From f278bd1a47f224f54597060d4ce108e585622ad3 Mon Sep 17 00:00:00 2001 From: qananasikq Date: Mon, 27 Apr 2026 13:09:45 +0300 Subject: [PATCH] Add worker fast sync module --- iaai_scraper/worker/fast_sync.py | 472 +++++++++++++++++++++++++++++++ 1 file changed, 472 insertions(+) create mode 100644 iaai_scraper/worker/fast_sync.py diff --git a/iaai_scraper/worker/fast_sync.py b/iaai_scraper/worker/fast_sync.py new file mode 100644 index 0000000..92911ce --- /dev/null +++ b/iaai_scraper/worker/fast_sync.py @@ -0,0 +1,472 @@ +from __future__ import annotations + +import concurrent.futures +import logging +import random +import time +from dataclasses import dataclass +from typing import Any, Callable + +from .browser.fast_client import FastListingVehicle, IAAIFastClient +from .core.runtime_config import RuntimeConfig +from .parsing.fast_mapper import INACTIVE_STATUS_VALUES, FastCarMapper +from .storage.db import PersistenceService +from .storage.schemas import CarRecord + +logger = logging.getLogger("iaai_scraper.fast_sync") + + +@dataclass(slots=True) +class FastSyncStats: + ids_fetched: int = 0 + cars_upserted: int = 0 + cars_failed: int = 0 + cars_filtered: int = 0 + images_upserted: int = 0 + skipped_existing: int = 0 + protection_events: int = 0 + + +def passes_condition_check(row: FastListingVehicle) -> bool: + if row.timed_auction_closed: + return False + status = (row.inventory_status or "").strip().upper() + return status not in INACTIVE_STATUS_VALUES + + +class FastSyncEngine: + """Full iaai-fast style sync pipeline adapted to this project's storage. + + Flow: hidden listing payloads -> concurrent ProductDetailsVM HTTP fetch -> + CarRecord preparation in memory -> single batch DB upsert. Browser is used + only inside IAAIFastClient to refresh cookies when IAAI challenge appears. + """ + + def __init__( + self, + *, + client: IAAIFastClient, + mapper: FastCarMapper, + persistence: PersistenceService, + batch_size: int, + fetch_concurrency: int, + report_progress: Callable[[str, Any], None] | None = None, + ) -> None: + self.client = client + self.mapper = mapper + self.persistence = persistence + self.batch_size = max(1, int(batch_size)) + self.fetch_concurrency = max(1, int(fetch_concurrency)) + self.report_progress = report_progress + + @staticmethod + def _is_transient_detail_error(exc: Exception) -> bool: + text = str(exc).lower() + return any( + marker in text + for marker in ( + "sslerror", + "ssleoferror", + "unexpected_eof_while_reading", + "eof occurred in violation of protocol", + "max retries exceeded", + "read timed out", + "readtimeout", + "connection reset", + "connection aborted", + "connection closed", + "temporarily unavailable", + "too many requests", + "status=429", + "status=500", + "status=502", + "status=503", + "status=504", + ) + ) + + def sync_listing( + self, + *, + runtime_config: RuntimeConfig, + make: str | None = None, + model: str | None = None, + lane: str = "iaai_cars", + limit: int | None = None, + only_new: bool = False, + listing_url: str | None = None, + max_pages: int | None = None, + skip_mark_sold: bool = False, + ) -> dict[str, Any]: + del lane + started_at = time.perf_counter() + stats = FastSyncStats() + errors: list[dict[str, str]] = [] + selected: dict[str, FastListingVehicle] = {} + rows_seen = 0 + rows_skipped_condition = 0 + + filters = runtime_config.filters + logger.info( + "Fast HTTP-first listing started: make=%s listing_url=%s max_pages=%s concurrency=%s batch_size=%s", + make or "ALL", + listing_url or "default", + max_pages, + self.fetch_concurrency, + self.batch_size, + ) + for vehicle in self.client.iter_listing_vehicles( + listing_start_url=listing_url, + make=make, + max_pages=max_pages, + ): + rows_seen += 1 + if runtime_config.sync.condition_check_enabled and not passes_condition_check(vehicle): + rows_skipped_condition += 1 + continue + if vehicle.inventory_id in selected: + continue + selected[vehicle.inventory_id] = vehicle + if limit is not None and limit > 0 and len(selected) >= limit: + break + + candidates = list(selected.values()) + all_listing_origin_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates} + if only_new and candidates: + origin_urls = [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates] + origin_ids = [f"iaai:{v.inventory_id}" for v in candidates] + existing_urls, existing_ids = self.persistence.get_existing_urls_and_ids(origin_urls, origin_ids) + fresh: list[FastListingVehicle] = [] + for vehicle in candidates: + if f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}" in existing_urls or f"iaai:{vehicle.inventory_id}" in existing_ids: + stats.skipped_existing += 1 + continue + fresh.append(vehicle) + candidates = fresh + + self._progress( + "fast_listing_collected", + rows_seen=rows_seen, + rows_filtered_condition=rows_skipped_condition, + rows_selected=len(candidates), + skipped_existing=stats.skipped_existing, + ) + logger.info( + "Fast HTTP-first listing collected: rows_seen=%d selected=%d skipped_existing=%d filtered_condition=%d only_new=%s", + rows_seen, + len(candidates), + stats.skipped_existing, + rows_skipped_condition, + only_new, + ) + + scan_completed = not (limit is not None and limit > 0) and not errors + + if not candidates: + return self._result( + started_at=started_at, + stats=stats, + failures=errors, + listing={ + "mode": "fast_hidden_payload", + "vehicles_collected": 0, + "vehicle_urls": [], + "early_stopped": False, + "truncated_by_time_budget": False, + "rows_seen": rows_seen, + "rows_filtered_condition": rows_skipped_condition, + }, + all_listing_origin_urls=all_listing_origin_urls, + full_scan_completed=scan_completed, + ) + + prepared_rows: list[CarRecord] = [] + db_processed = 0 + retry_candidates: list[FastListingVehicle] = [] + started_details = time.perf_counter() + with concurrent.futures.ThreadPoolExecutor(max_workers=min(self.fetch_concurrency, len(candidates))) as executor: + future_to_vehicle = { + executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle + for vehicle in candidates + } + for index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1): + vehicle = future_to_vehicle[future] + try: + payload = future.result() + record = self.mapper.map_payload_to_record( + detail_payload=payload, + vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", + listing_vehicle=vehicle, + ) + if model and model.casefold() not in record.model.casefold(): + stats.cars_filtered += 1 + continue + if not filters.matches({ + "brand": record.brand, + "model": record.model, + "year": record.year, + "body_type": record.body_type, + "color": record.color, + "drive": record.drive, + "gearbox": record.gearbox, + "price": record.price, + "mileage": record.mileage, + }): + stats.cars_filtered += 1 + continue + prepared_rows.append(record) + stats.ids_fetched += 1 + if len(prepared_rows) >= self.batch_size: + db_processed += self._flush_db_records( + rows=prepared_rows, + stats=stats, + errors=errors, + processed=db_processed + len(prepared_rows), + total=len(candidates), + ) + prepared_rows.clear() + except Exception as exc: + stats.cars_failed += 1 + errors.append({"vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", "error": str(exc)}) + if _looks_like_protection(exc): + stats.protection_events += 1 + if self._is_transient_detail_error(exc): + retry_candidates.append(vehicle) + logger.info("Fast detail transient failure queued for retry inventory_id=%s: %s", vehicle.inventory_id, exc) + else: + logger.exception("Fast detail parse failed inventory_id=%s: %s", vehicle.inventory_id, exc) + + if index % 100 == 0 or index == len(candidates): + elapsed = max(0.001, time.perf_counter() - started_details) + if self.client._settings.scraping_profile.verbose_progress_logs: + logger.info( + "Fast HTTP-first details progress: processed=%d/%d ok=%d failed=%d queued_db=%d rate=%.2f/s", + index, + len(candidates), + stats.ids_fetched, + stats.cars_failed, + len(prepared_rows), + index / elapsed, + ) + self._progress( + "fast_detail_progress", + processed=index, + total=len(candidates), + ids_fetched=stats.ids_fetched, + cars_failed=stats.cars_failed, + queued_for_db=len(prepared_rows), + throughput=round(index / elapsed, 2), + ) + + if retry_candidates: + retry_started = time.perf_counter() + retry_workers = max(1, min(8, self.fetch_concurrency // 2, len(retry_candidates))) + retry_errors: list[dict[str, str]] = [] + logger.info( + "Fast HTTP-first retrying transient detail failures: total=%d workers=%d", + len(retry_candidates), + retry_workers, + ) + with concurrent.futures.ThreadPoolExecutor(max_workers=retry_workers) as executor: + future_to_vehicle = { + executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle + for vehicle in retry_candidates + } + for retry_index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1): + vehicle = future_to_vehicle[future] + try: + payload = future.result() + record = self.mapper.map_payload_to_record( + detail_payload=payload, + vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", + listing_vehicle=vehicle, + ) + if model and model.casefold() not in record.model.casefold(): + stats.cars_filtered += 1 + continue + if not filters.matches({ + "brand": record.brand, + "model": record.model, + "year": record.year, + "body_type": record.body_type, + "color": record.color, + "drive": record.drive, + "gearbox": record.gearbox, + "price": record.price, + "mileage": record.mileage, + }): + stats.cars_filtered += 1 + continue + prepared_rows.append(record) + stats.ids_fetched += 1 + stats.cars_failed = max(0, stats.cars_failed - 1) + if len(prepared_rows) >= self.batch_size: + db_processed += self._flush_db_records( + rows=prepared_rows, + stats=stats, + errors=errors, + processed=db_processed + len(prepared_rows), + total=len(candidates), + ) + prepared_rows.clear() + except Exception as exc: + retry_errors.append({ + "vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", + "error": str(exc), + }) + if _looks_like_protection(exc): + stats.protection_events += 1 + if retry_index % 100 == 0 or retry_index == len(retry_candidates): + elapsed = max(0.001, time.perf_counter() - retry_started) + logger.info( + "Fast HTTP-first retry progress: processed=%d/%d recovered=%d remaining_failed=%d rate=%.2f/s", + retry_index, + len(retry_candidates), + len(retry_candidates) - len(retry_errors), + len(retry_errors), + retry_index / elapsed, + ) + transient_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in retry_candidates} + errors = [error for error in errors if error.get("vehicle_url") not in transient_urls] + errors.extend(retry_errors) + + if prepared_rows: + db_processed += self._flush_db_records( + rows=prepared_rows, + stats=stats, + errors=errors, + processed=db_processed + len(prepared_rows), + total=len(candidates), + ) + prepared_rows.clear() + + self.client.persist_session_state() + + scan_completed = not (limit is not None and limit > 0) and not errors + mark_sold_scope_partial = bool( + (limit is not None and limit > 0) + or make + or model + or listing_url + or only_new + ) + if all_listing_origin_urls and not mark_sold_scope_partial and not skip_mark_sold and scan_completed: + try: + sold_count = self.persistence.mark_sold_not_in_listing_by_urls(all_listing_origin_urls, lane="iaai") + except Exception as exc: + sold_count = 0 + logger.warning("Fast sold reconcile failed: %s", exc) + else: + sold_count = 0 + + listing = { + "mode": "fast_hidden_payload", + "vehicles_collected": len(candidates), + "vehicle_urls": [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates], + "early_stopped": False, + "truncated_by_time_budget": False, + "rows_seen": rows_seen, + "rows_filtered_condition": rows_skipped_condition, + "sold_marked": sold_count, + } + return self._result( + started_at=started_at, + stats=stats, + failures=errors, + listing=listing, + all_listing_origin_urls=all_listing_origin_urls, + full_scan_completed=scan_completed, + ) + + def _result( + self, + *, + started_at: float, + stats: FastSyncStats, + failures: list[dict[str, str]], + listing: dict[str, Any], + all_listing_origin_urls: set[str], + full_scan_completed: bool, + ) -> dict[str, Any]: + status = "success" if not failures else ("partial_success" if stats.cars_upserted else "failed") + total = int(listing.get("vehicles_collected") or 0) + fail_ratio = (stats.cars_failed / total) if total > 0 else 0.0 + protection_ratio = (stats.protection_events / total) if total > 0 else 0.0 + anti_bot_detected = total > 0 and ((stats.protection_events >= 30 and protection_ratio >= 0.10) or fail_ratio >= 0.30) + return { + "status": status, + "listing": listing, + "total": total, + "total_discovered": total, + "skipped_existing": stats.skipped_existing, + "cars_upserted": stats.cars_upserted, + "cars_failed": stats.cars_failed, + "cars_filtered": stats.cars_filtered, + "images_upserted": stats.images_upserted, + "protection_events": stats.protection_events, + "failures": failures, + "all_listing_origin_urls": all_listing_origin_urls, + "full_scan_completed": full_scan_completed and not anti_bot_detected, + "anti_bot_detected": anti_bot_detected, + "fail_ratio": round(fail_ratio, 4), + "protection_ratio": round(protection_ratio, 4), + "elapsed_seconds": round(time.perf_counter() - started_at, 3), + } + + def _progress(self, stage: str, **meta: Any) -> None: + if self.report_progress is None: + return + try: + self.report_progress(stage, **meta) + except Exception: + pass + + def _fetch_detail_payload(self, inventory_id: str) -> dict[str, Any]: + jitter = float(self.client._settings.scraping_profile.request_jitter_max_s) + if jitter > 0: + time.sleep(random.uniform(0.0, jitter)) + return self.client.fetch_vehicle_detail_payload(inventory_id) + + def _flush_db_records( + self, + *, + rows: list[CarRecord], + stats: FastSyncStats, + errors: list[dict[str, str]], + processed: int, + total: int, + ) -> int: + if not rows: + return 0 + batch = list(rows) + try: + upsert = self.persistence.upsert_cars_batch(batch) + stats.cars_upserted += int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0)) + stats.images_upserted += int(upsert.get("images_upserted", 0)) + except Exception as exc: + stats.cars_failed += len(batch) + errors.append({"vehicle_url": f"db_batch_{processed - len(batch)}", "error": str(exc)}) + logger.exception("Fast DB apply failed processed=%s size=%s: %s", processed, len(batch), exc) + self._progress( + "fast_db_progress", + processed=processed, + total=total, + cars_upserted=stats.cars_upserted, + cars_failed=stats.cars_failed, + images_upserted=stats.images_upserted, + ) + logger.info( + "Fast HTTP-first DB batch: processed=%d/%d batch=%d upserted=%d failed=%d images=%d", + processed, + total, + len(batch), + stats.cars_upserted, + stats.cars_failed, + stats.images_upserted, + ) + return len(batch) + + +def _looks_like_protection(exc: Exception) -> bool: + message = str(exc).lower() + return any(token in message for token in ("captcha", "antibot", "challenge", "blocked", "403", "429", "incapsula"))