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) verbose_progress_logs = bool( getattr( getattr(getattr(self.client, "_settings", None), "scraping_profile", None), "verbose_progress_logs", False, ) ) if 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"))