487 lines
20 KiB
Python
487 lines
20 KiB
Python
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] = []
|
|
transient_failure_log_count = 0
|
|
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)
|
|
transient_failure_log_count += 1
|
|
if transient_failure_log_count % 100 == 0:
|
|
logger.warning(
|
|
"Fast detail transient failures queued for retry: count=%d latest_inventory_id=%s",
|
|
transient_failure_log_count,
|
|
vehicle.inventory_id,
|
|
)
|
|
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"))
|