stabilize daemon sync and docker startup

2026-04-08 14:30:44 +03:00
parent d1844ba625
commit b2d3902bcc
4 changed files with 79 additions and 42 deletions
--- a/iaai_scraper/cli.py
+++ b/iaai_scraper/cli.py
@@ -1,6 +1,7 @@
 import argparse
 from pathlib import Path

+from .core.config import Settings
 from .core.utils import save_to_json
 from .scraper import IAAIScraper

@@ -75,8 +76,6 @@ def main() -> None:

    runtime_settings: Settings | None = None
    if args.headless is not None or args.debug:
-        from .core.config import Settings
-
        runtime_settings = Settings()
        if args.headless is not None:
            runtime_settings.headless = args.headless == "true"
@@ -85,7 +84,6 @@ def main() -> None:

    if args.command == "run-daemon":
        # daemon: бесконечный цикл
-        from .core.config import Settings
        runtime_settings = runtime_settings or Settings()
        if args.interval is not None:
            runtime_settings.scheduler_interval_minutes = args.interval
--- a/iaai_scraper/scraper.py
+++ b/iaai_scraper/scraper.py
@@ -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