From 69056086cc6a03a40b360bbc8916a23974330894 Mon Sep 17 00:00:00 2001 From: qananasikq Date: Tue, 28 Apr 2026 22:14:13 +0300 Subject: [PATCH] Fast batch --- iaai_scraper/mobile_de/scraper.py | 261 +++++++++++++++++++++--------- 1 file changed, 185 insertions(+), 76 deletions(-) diff --git a/iaai_scraper/mobile_de/scraper.py b/iaai_scraper/mobile_de/scraper.py index 3135a36..87d8b41 100644 --- a/iaai_scraper/mobile_de/scraper.py +++ b/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) + else: + existing_streak = 0 + new_records_kept += 1 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, - ) 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)