update sync
This commit is contained in:
qananasikq
@@ -175,6 +175,7 @@ class FastSyncEngine:
|
||||
)
|
||||
|
||||
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:
|
||||
@@ -209,6 +210,15 @@ class FastSyncEngine:
|
||||
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)})
|
||||
@@ -222,15 +232,16 @@ class FastSyncEngine:
|
||||
|
||||
if index % 100 == 0 or index == len(candidates):
|
||||
elapsed = max(0.001, time.perf_counter() - started_details)
|
||||
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,
|
||||
)
|
||||
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,
|
||||
@@ -283,6 +294,15 @@ class FastSyncEngine:
|
||||
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}",
|
||||
@@ -305,32 +325,14 @@ class FastSyncEngine:
|
||||
errors.extend(retry_errors)
|
||||
|
||||
if prepared_rows:
|
||||
for batch_start in range(0, len(prepared_rows), self.batch_size):
|
||||
batch = prepared_rows[batch_start:batch_start + self.batch_size]
|
||||
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_{batch_start}", "error": str(exc)})
|
||||
logger.exception("Fast DB apply failed batch_start=%s size=%s: %s", batch_start, len(batch), exc)
|
||||
self._progress(
|
||||
"fast_db_progress",
|
||||
processed=min(batch_start + len(batch), len(prepared_rows)),
|
||||
total=len(prepared_rows),
|
||||
cars_upserted=stats.cars_upserted,
|
||||
cars_failed=stats.cars_failed,
|
||||
images_upserted=stats.images_upserted,
|
||||
)
|
||||
logger.info(
|
||||
"Fast HTTP-first DB progress: processed=%d/%d upserted=%d failed=%d images=%d",
|
||||
min(batch_start + len(batch), len(prepared_rows)),
|
||||
len(prepared_rows),
|
||||
stats.cars_upserted,
|
||||
stats.cars_failed,
|
||||
stats.images_upserted,
|
||||
)
|
||||
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()
|
||||
|
||||
@@ -419,6 +421,45 @@ class FastSyncEngine:
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user