Fast batch

iaai_scraper/mobile_de/scraper.py

@@ -8,6 +8,7 @@ from typing import Any
 
 from ..core.config import Settings, settings
 from ..storage.db import PersistenceService
+from ..storage.schemas import CarRecord
 from .client import MobileDeClient
 from .mapper import MobileDeMapper
 from .models import MobileDeListing
@@ -16,6 +17,10 @@ logger = logging.getLogger("mobile_de.scraper")
 
 MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK", "2")))
 MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS", "1")))
+MOBILEDE_SEARCH_STRATEGY_NOTE = (
+    "mobile.de search pages return about 20 listings per page and are limited to about 50 pages; "
+    "for full coverage split into narrower segments and deduplicate by id."
+)
 
 
 class MobileDeScraper:
@@ -33,6 +38,102 @@ class MobileDeScraper:
         self.persistence = persistence or PersistenceService(self.settings)
         self.mapper = mapper or MobileDeMapper()
 
+    @staticmethod
+    def _should_cut_only_new_tail(sort_by: str | None, sort_order: str | None) -> bool:
+        return (
+            str(sort_by or "").lower() == "doc"
+            and str(sort_order or "").lower() == "down"
+            and MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK > 0
+        )
+
+    def _build_search_params(
+        self,
+        *,
+        search_url: str | None,
+        make_id: str | None,
+        model_id: str | None,
+        price_min: str | None,
+        price_max: str | None,
+        year_min: str | None,
+        year_max: str | None,
+        mileage_min: str | None,
+        mileage_max: str | None,
+        sort_by: str | None,
+        sort_order: str | None,
+    ) -> dict[str, str | None]:
+        params: dict[str, str | 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:
+            params["sb"] = sort_by
+        if sort_order:
+            params["od"] = sort_order
+        return params
+
+    @staticmethod
+    def _dedupe_page_records(
+        page_records: list[CarRecord],
+        seen_record_keys: set[str],
+    ) -> list[CarRecord]:
+        deduped_page_records: list[CarRecord] = []
+        for record in page_records:
+            record_key = record.origin_id or record.origin_url
+            if not record_key or record_key in seen_record_keys:
+                continue
+            seen_record_keys.add(record_key)
+            deduped_page_records.append(record)
+        return deduped_page_records
+
+    def _apply_only_new_page_policy(
+        self,
+        *,
+        page_records: list[CarRecord],
+        only_new: bool | None,
+        sort_by: str | None,
+        sort_order: str | None,
+        skipped_existing: int,
+        existing_streak: int,
+        new_records_kept: int,
+    ) -> tuple[list[CarRecord], int, int, int, bool]:
+        if not only_new or not page_records:
+            return page_records, skipped_existing, existing_streak, new_records_kept, False
+
+        existing_origin_ids = self.persistence.get_existing_origin_ids(
+            [record.origin_id for record in page_records if record.origin_id]
+        )
+        head_cut_triggered = False
+        should_cut_tail = self._should_cut_only_new_tail(sort_by, sort_order)
+
+        if should_cut_tail:
+            filtered_records: list[CarRecord] = []
+            for record_index, record in enumerate(page_records):
+                is_existing = bool(record.origin_id and record.origin_id in existing_origin_ids)
+                if is_existing:
+                    existing_streak += 1
+                    if (
+                        existing_streak >= MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK
+                        and new_records_kept >= MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS
+                    ):
+                        head_cut_triggered = True
+                        skipped_existing += max(0, len(page_records) - record_index - 1)
+                        break
+                else:
+                    existing_streak = 0
+                    new_records_kept += 1
+                filtered_records.append(record)
+            return filtered_records, skipped_existing, existing_streak, new_records_kept, head_cut_triggered
+
+        for record in page_records:
+            if not record.origin_id or record.origin_id not in existing_origin_ids:
+                new_records_kept += 1
+        return page_records, skipped_existing, existing_streak, new_records_kept, False
+
     def collect_search(
         self,
         *,
@@ -51,19 +152,19 @@ class MobileDeScraper:
         sort_order: str | None = None,
         progress_callback: Callable[[str, dict[str, Any]], None] | None = None,
     ) -> dict[str, Any]:
-        params: dict[str, str | 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:
-            params["sb"] = sort_by
-        if sort_order:
-            params["od"] = sort_order
+        params = self._build_search_params(
+            search_url=search_url,
+            make_id=make_id,
+            model_id=model_id,
+            price_min=price_min,
+            price_max=price_max,
+            year_min=year_min,
+            year_max=year_max,
+            mileage_min=mileage_min,
+            mileage_max=mileage_max,
+            sort_by=sort_by,
+            sort_order=sort_order,
+        )
 
         logger.debug(
             "mobile.de collect_search started: start_page=%s max_pages=%s search_url=%s make_id=%s model_id=%s year=%s-%s price=%s-%s mileage=%s-%s",
@@ -114,10 +215,7 @@ class MobileDeScraper:
         unique_ids = sorted({listing.id for page in pages for listing in page.listings if listing.id})
         result = {
             "source": "mobile.de",
-            "strategy_note": (
-                "mobile.de UI shows 50 pages, but pageNumber works beyond 50; "
-                "for full coverage split by make/model/year/price and deduplicate by id."
-            ),
+            "strategy_note": MOBILEDE_SEARCH_STRATEGY_NOTE,
             "search_url": search_url,
             "pages": [asdict(page) for page in pages],
             "listing_count": sum(len(page.listings) for page in pages),
@@ -175,7 +273,19 @@ class MobileDeScraper:
     ) -> dict[str, Any]:
         self.persistence.create_tables()
         run_id = self.persistence.start_sync_run(lane)
-        failed = 0
+        pages_payload: list[dict[str, Any]] = []
+        unique_ids: set[str] = set()
+        listing_count = 0
+        skipped_existing = 0
+        ids_fetched = 0
+        cars_upserted = 0
+        inserted_total = 0
+        updated_total = 0
+        images_upserted = 0
+        existing_streak = 0
+        new_records_kept = 0
+        head_cut_triggered = False
+        seen_record_keys: set[str] = set()
         logger.debug(
             "mobile.de sync_search started: run_id=%s lane=%s start_page=%s max_pages=%s",
             run_id,
@@ -200,61 +310,40 @@ class MobileDeScraper:
                 sort_order=sort_order,
                 progress_callback=progress_callback,
             )
-            # Map serialized listings back to records.
-            records = []
-            for page in data["pages"]:
-                for item in page["listings"]:
-                    records.append(self.mapper.listing_to_car_record(MobileDeListing(**item)))
 
-            skipped_existing = 0
+            pages_payload = list(data.get("pages", []))
+            listing_count = int(data.get("listing_count", 0) or 0)
+            unique_ids = set(data.get("unique_listing_ids", []))
+            pages_collected = len(pages_payload)
+
+            records: list[CarRecord] = []
+            for page in pages_payload:
+                page_records = [self.mapper.listing_to_car_record(MobileDeListing(**item)) for item in page.get("listings", [])]
+                records.extend(self._dedupe_page_records(page_records, seen_record_keys))
+
             if only_new and records:
                 existing_origin_ids = self.persistence.get_existing_origin_ids(
                     [record.origin_id for record in records if record.origin_id]
                 )
-                before_filter_count = len(records)
-
-                # For newest-first, cut tail after existing streak.
-                should_cut_tail = (
-                    str(sort_by or "").lower() == "doc"
-                    and str(sort_order or "").lower() == "down"
-                    and MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK > 0
-                )
+                should_cut_tail = self._should_cut_only_new_tail(sort_by, sort_order)
                 if should_cut_tail:
-                    filtered_records = []
-                    existing_streak = 0
-                    considered = 0
-                    for record in records:
-                        considered += 1
-                        is_existing = record.origin_id in existing_origin_ids
+                    filtered_records: list[CarRecord] = []
+                    for record_index, record in enumerate(records):
+                        is_existing = bool(record.origin_id and record.origin_id in existing_origin_ids)
+                        filtered_records.append(record)
                         if is_existing:
-                            skipped_existing += 1
                             existing_streak += 1
                             if (
                                 existing_streak >= MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK
-                                and len(filtered_records) >= MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS
+                                and new_records_kept >= MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS
                             ):
+                                head_cut_triggered = True
+                                skipped_existing += max(0, len(records) - record_index - 1)
                                 break
-                            continue
-                        existing_streak = 0
-                        filtered_records.append(record)
-                    records = filtered_records
-                    data["listing_count"] = considered
-                    data["unique_listing_count"] = min(int(data.get("unique_listing_count", considered)), considered)
-                    logger.info(
-                        "mobile.de only_new head-cut applied: considered=%s kept=%s skipped_existing=%s streak=%s",
-                        considered,
-                        len(records),
-                        skipped_existing,
-                        MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK,
-                    )
                         else:
-                    records = [record for record in records if record.origin_id not in existing_origin_ids]
-                    skipped_existing = before_filter_count - len(records)
-                logger.info(
-                    "mobile.de only_new filter applied: kept=%s skipped_existing=%s",
-                    len(records),
-                    skipped_existing,
-                )
+                            existing_streak = 0
+                            new_records_kept += 1
+                    records = filtered_records
 
             if progress_callback is not None:
                 progress_callback(
@@ -264,6 +353,7 @@ class MobileDeScraper:
                         "skipped_existing": skipped_existing,
                         "only_new": bool(only_new),
                         "run_id": run_id,
+                        "pages_collected": pages_collected,
                     },
                 )
 
@@ -272,30 +362,49 @@ class MobileDeScraper:
                 "updated": 0,
                 "images_upserted": 0,
             }
+            ids_fetched = len(records)
+            inserted_total = int(upsert.get("inserted", 0))
+            updated_total = int(upsert.get("updated", 0))
+            images_upserted = int(upsert.get("images_upserted", 0))
+            cars_upserted = inserted_total + updated_total
+
             logger.debug(
-                "mobile.de sync_search upsert finished: run_id=%s inserted=%s updated=%s images=%s",
+                "mobile.de sync_search batch upsert: run_id=%s pages=%s inserted=%s updated=%s images=%s",
                 run_id,
-                upsert.get("inserted", 0),
-                upsert.get("updated", 0),
-                upsert.get("images_upserted", 0),
+                pages_collected,
+                inserted_total,
+                updated_total,
+                images_upserted,
             )
             if progress_callback is not None:
                 progress_callback(
                     "db_upsert_done",
                     {
                         "run_id": run_id,
-                        "inserted": int(upsert.get("inserted", 0)),
-                        "updated": int(upsert.get("updated", 0)),
-                        "images_upserted": int(upsert.get("images_upserted", 0)),
+                        "pages_collected": pages_collected,
+                        "pages_in_batch": pages_collected,
+                        "inserted": inserted_total,
+                        "updated": updated_total,
+                        "images_upserted": images_upserted,
                     },
                 )
+
+            if head_cut_triggered:
+                logger.info(
+                    "mobile.de only_new head-cut applied: kept=%s skipped_existing=%s streak=%s",
+                    len(records),
+                    skipped_existing,
+                    MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK,
+                )
+
+            data["early_stopped"] = head_cut_triggered
             self.persistence.finish_sync_run(
                 run_id,
                 status="success",
-                ids_fetched=len(records),
-                cars_upserted=int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0)),
-                cars_failed=failed,
-                images_upserted=int(upsert.get("images_upserted", 0)),
+                ids_fetched=ids_fetched,
+                cars_upserted=cars_upserted,
+                cars_failed=0,
+                images_upserted=images_upserted,
             )
             logger.debug(
                 "mobile.de sync_search completed: run_id=%s listings=%s unique=%s",
@@ -303,15 +412,15 @@ class MobileDeScraper:
                 data.get("listing_count", 0),
                 data.get("unique_listing_count", 0),
             )
-            return {"run_id": run_id, "source": "mobile.de", "upsert": upsert, "skipped_existing": skipped_existing, **data}
+            return {"run_id": run_id, "upsert": upsert, "skipped_existing": skipped_existing, **data}
         except Exception as exc:
             self.persistence.finish_sync_run(
                 run_id,
                 status="failed",
-                ids_fetched=0,
-                cars_upserted=0,
-                cars_failed=failed,
-                images_upserted=0,
+                ids_fetched=ids_fetched,
+                cars_upserted=cars_upserted,
+                cars_failed=0,
+                images_upserted=images_upserted,
                 error_summary=str(exc),
             )
             logger.error("mobile.de sync_search failed: run_id=%s error=%s", run_id, exc, exc_info=True)