improve scraper runtime
This commit is contained in:
qananasikq
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
import signal
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
@@ -39,6 +40,7 @@ class IAAIScraper:
|
||||
self.vehicle_parser = VehicleParser()
|
||||
self.car_mapper = CarMapper()
|
||||
self.persistence = PersistenceService(self.settings)
|
||||
self._shutdown_requested = False
|
||||
|
||||
# browser lifecycle
|
||||
|
||||
@@ -170,7 +172,7 @@ class IAAIScraper:
|
||||
trace_id = self._new_trace_id("scrape")
|
||||
started_at = time.perf_counter()
|
||||
capture = NetworkCapture(self.settings)
|
||||
capture.attach(page)
|
||||
capture.attach(page, origin_url=vehicle_url)
|
||||
|
||||
page.goto(vehicle_url, wait_until="domcontentloaded", timeout=60_000)
|
||||
try:
|
||||
@@ -264,24 +266,26 @@ class IAAIScraper:
|
||||
trace_id = self._new_trace_id("sync-listing")
|
||||
started_at = time.perf_counter()
|
||||
self.persistence.create_tables()
|
||||
listing = self.collect_listing(make=make, model=model)
|
||||
vehicle_urls = list(listing.get("vehicle_urls", []))
|
||||
if limit is not None:
|
||||
vehicle_urls = vehicle_urls[:max(0, limit)]
|
||||
run_id = self.persistence.start_sync_run(lane=lane)
|
||||
cars_upserted = 0
|
||||
cars_failed = 0
|
||||
images_upserted = 0
|
||||
total = 0
|
||||
failures: list[dict[str, str]] = []
|
||||
listing: dict = {}
|
||||
|
||||
total = len(vehicle_urls)
|
||||
logger.info("Starting sync: %d vehicles to process", total)
|
||||
|
||||
page = self._get_page()
|
||||
try:
|
||||
listing = self.collect_listing(make=make, model=model)
|
||||
vehicle_urls = list(listing.get("vehicle_urls", []))
|
||||
if limit is not None:
|
||||
vehicle_urls = vehicle_urls[:max(0, limit)]
|
||||
|
||||
total = len(vehicle_urls)
|
||||
logger.info("Starting sync: %d vehicles to process", total)
|
||||
|
||||
for index, vehicle_url in enumerate(vehicle_urls, start=1):
|
||||
page = self._get_page()
|
||||
try:
|
||||
# Повторно используем страницу, чтобы не создавать лишний overhead.
|
||||
logger.info("[%d/%d] Scraping %s", index, total, vehicle_url)
|
||||
scrape_result = self._scrape_on_page(page, vehicle_url)
|
||||
db_record = scrape_result.get("db_record")
|
||||
@@ -289,7 +293,8 @@ class IAAIScraper:
|
||||
raise RuntimeError("Scrape result does not contain db_record")
|
||||
record = CarRecord.model_validate(db_record)
|
||||
upsert = self.persistence.upsert_car(record)
|
||||
cars_upserted += 1
|
||||
if upsert.get("action") != "skipped":
|
||||
cars_upserted += 1
|
||||
img_count = int(upsert.get("images_upserted", 0))
|
||||
images_upserted += img_count
|
||||
logger.info(
|
||||
@@ -299,35 +304,35 @@ class IAAIScraper:
|
||||
record.year or "?", record.price or "N/A", img_count,
|
||||
)
|
||||
except Exception as exc:
|
||||
# После сбоя пересоздаём page, чтобы не остаться в битом состоянии.
|
||||
cars_failed += 1
|
||||
failures.append({"vehicle_url": vehicle_url, "error": str(exc)})
|
||||
logger.error("[%d/%d] Failed %s: %s", index, total, vehicle_url, exc)
|
||||
finally:
|
||||
try:
|
||||
page.close()
|
||||
except Exception:
|
||||
pass
|
||||
page = self._get_page()
|
||||
|
||||
if index < total:
|
||||
self.pacer.between_vehicles()
|
||||
except Exception as exc:
|
||||
# collect_listing itself failed — count as total failure
|
||||
if not failures:
|
||||
failures.append({"vehicle_url": "collect_listing", "error": str(exc)})
|
||||
logger.error("sync_listing failed: %s", exc)
|
||||
finally:
|
||||
try:
|
||||
page.close()
|
||||
except Exception:
|
||||
pass
|
||||
status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
|
||||
error_summary = "; ".join(item["error"] for item in failures[:10]) if failures else None
|
||||
self.persistence.finish_sync_run(
|
||||
run_id,
|
||||
status=status,
|
||||
ids_fetched=total,
|
||||
cars_upserted=cars_upserted,
|
||||
cars_failed=cars_failed,
|
||||
images_upserted=images_upserted,
|
||||
error_summary=error_summary,
|
||||
)
|
||||
|
||||
status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
|
||||
error_summary = "; ".join(item["error"] for item in failures[:10]) if failures else None
|
||||
self.persistence.finish_sync_run(
|
||||
run_id,
|
||||
status=status,
|
||||
ids_fetched=total,
|
||||
cars_upserted=cars_upserted,
|
||||
cars_failed=cars_failed,
|
||||
images_upserted=images_upserted,
|
||||
error_summary=error_summary,
|
||||
)
|
||||
logger.info(
|
||||
"Sync run #%d finished: %d/%d upserted, %d failed, %d images",
|
||||
run_id, cars_upserted, total, cars_failed, images_upserted,
|
||||
@@ -348,17 +353,26 @@ class IAAIScraper:
|
||||
|
||||
def run_scheduled(self) -> None:
|
||||
# Простой бесконечный цикл без внешнего планировщика.
|
||||
# Graceful shutdown по SIGINT/SIGTERM.
|
||||
def _handle_shutdown(signum, frame):
|
||||
logger.info("Received signal %s, shutting down gracefully...", signum)
|
||||
self._shutdown_requested = True
|
||||
|
||||
signal.signal(signal.SIGINT, _handle_shutdown)
|
||||
signal.signal(signal.SIGTERM, _handle_shutdown)
|
||||
|
||||
interval = self.settings.scheduler_interval_minutes * 60
|
||||
logger.info(
|
||||
"Scheduler started: syncing every %d minutes",
|
||||
self.settings.scheduler_interval_minutes,
|
||||
)
|
||||
cycle = 0
|
||||
while True:
|
||||
while not self._shutdown_requested:
|
||||
cycle += 1
|
||||
logger.info("=== Scheduler cycle #%d starting ===", cycle)
|
||||
start = time.time()
|
||||
try:
|
||||
# Сбрасываем контекст перед каждым циклом.
|
||||
if self.context:
|
||||
try:
|
||||
self.context.close()
|
||||
@@ -366,6 +380,12 @@ class IAAIScraper:
|
||||
pass
|
||||
self.context = None
|
||||
|
||||
# Проверяем, что browser/playwright живы; пересоздаём при необходимости.
|
||||
if self.browser is None or self.playwright is None:
|
||||
logger.info("Browser/Playwright not available, re-initializing...")
|
||||
self.close()
|
||||
self.__enter__()
|
||||
|
||||
result = self.sync_listing()
|
||||
elapsed = time.time() - start
|
||||
logger.info(
|
||||
@@ -377,17 +397,31 @@ class IAAIScraper:
|
||||
except Exception as exc:
|
||||
elapsed = time.time() - start
|
||||
logger.error("Cycle #%d failed after %.1fs: %s", cycle, elapsed, exc)
|
||||
if self.context:
|
||||
try:
|
||||
self.context.close()
|
||||
except PlaywrightError:
|
||||
pass
|
||||
self.context = None
|
||||
# Полный сброс при любой ошибке цикла — следующий цикл пересоздаст всё.
|
||||
try:
|
||||
self.close()
|
||||
except Exception:
|
||||
pass
|
||||
# Пересоздаём browser для следующего цикла.
|
||||
try:
|
||||
self.__enter__()
|
||||
except Exception as reinit_exc:
|
||||
logger.error("Failed to re-initialize browser: %s", reinit_exc)
|
||||
|
||||
if self._shutdown_requested:
|
||||
break
|
||||
|
||||
sleep_time = max(0, interval - (time.time() - start))
|
||||
if sleep_time > 0:
|
||||
logger.info("Sleeping %.0f seconds until next cycle...", sleep_time)
|
||||
time.sleep(sleep_time)
|
||||
# Прерываемый sleep — проверяем shutdown каждые 5 секунд.
|
||||
slept = 0.0
|
||||
while slept < sleep_time and not self._shutdown_requested:
|
||||
chunk = min(5.0, sleep_time - slept)
|
||||
time.sleep(chunk)
|
||||
slept += chunk
|
||||
|
||||
logger.info("Scheduler stopped gracefully after %d cycles.", cycle)
|
||||
|
||||
# helpers
|
||||
|
||||
|
||||
Reference in New Issue
Block a user