Add checkpoint logic and tests

iaai_scraper/worker/tasks.py

@@ -1,6 +1,7 @@
 # Задачи Celery для синхронизации автомобилей и листинга IAAI.
 
 from concurrent.futures import ThreadPoolExecutor
+import json
 import logging
 from threading import Event, Thread
 import time
@@ -18,6 +19,7 @@ logger = logging.getLogger("iaai_scraper.worker.tasks")
 
 SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
 SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
+SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
 SYNC_LISTING_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task"
 
 
@@ -228,6 +230,55 @@ def _set_full_scan_done(redis_client: Redis, done: bool) -> None:
         logger.warning("Failed to persist full scan state", exc_info=True)
 
 
+def _load_sync_checkpoint(redis_client: Redis) -> dict[str, object] | None:
+    try:
+        raw = redis_client.get(SYNC_LISTING_CHECKPOINT_KEY)
+    except Exception:
+        logger.warning("Failed to read sync listing checkpoint", exc_info=True)
+        return None
+    if not raw:
+        return None
+    if isinstance(raw, str) and not raw.strip():
+        return None
+    try:
+        data = json.loads(str(raw))
+    except Exception:
+        logger.warning("Failed to decode sync listing checkpoint", exc_info=True)
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _save_sync_checkpoint(
+    redis_client: Redis,
+    *,
+    task_id: str,
+    page_number: int,
+    make: str | None,
+    model: str | None,
+    lane: str,
+) -> None:
+    payload = {
+        "status": "in_progress",
+        "task_id": task_id,
+        "last_successful_page": int(page_number),
+        "make": make,
+        "model": model,
+        "lane": lane,
+        "updated_at": int(time.time()),
+    }
+    try:
+        redis_client.set(SYNC_LISTING_CHECKPOINT_KEY, json.dumps(payload))
+    except Exception:
+        logger.warning("Failed to save sync listing checkpoint", exc_info=True)
+
+
+def _clear_sync_checkpoint(redis_client: Redis) -> None:
+    try:
+        redis_client.delete(SYNC_LISTING_CHECKPOINT_KEY)
+    except Exception:
+        logger.warning("Failed to clear sync listing checkpoint", exc_info=True)
+
+
 def _start_lock_heartbeat(
     redis_client: Redis,
     key: str,
@@ -350,6 +401,28 @@ def sync_listing_task(
         force_bootstrap_full_scan = not full_scan_done_before_run
         effective_limit = None if force_bootstrap_full_scan else limit
         effective_only_new = False if force_bootstrap_full_scan else only_new
+        checkpoint = _load_sync_checkpoint(redis_client)
+        resume_from_page = 1
+
+        if checkpoint and str(checkpoint.get("status") or "") == "in_progress":
+            checkpoint_page = int(checkpoint.get("last_successful_page") or 0)
+            checkpoint_make = checkpoint.get("make")
+            checkpoint_model = checkpoint.get("model")
+            checkpoint_lane = checkpoint.get("lane")
+            same_scope = (
+                checkpoint_make == make
+                and checkpoint_model == model
+                and checkpoint_lane == lane
+            )
+            if checkpoint_page > 0 and same_scope:
+                resume_from_page = checkpoint_page + 1
+                logger.warning(
+                    "Resuming sync_listing from page %d using checkpoint",
+                    resume_from_page,
+                )
+            elif checkpoint_page > 0:
+                logger.info("Ignoring stale checkpoint due to different sync parameters")
+                _clear_sync_checkpoint(redis_client)
 
         if force_bootstrap_full_scan:
             logger.info(
@@ -377,6 +450,15 @@ def sync_listing_task(
                     lane=lane,
                     limit=effective_limit,
                     only_new=effective_only_new,
+                    start_page=resume_from_page,
+                    progress_callback=lambda page_number: _save_sync_checkpoint(
+                        redis_client,
+                        task_id=task_id,
+                        page_number=page_number,
+                        make=make,
+                        model=model,
+                        lane=lane,
+                    ),
                 )
 
         result = _run_browser_job(_job)
@@ -385,11 +467,14 @@ def sync_listing_task(
             bootstrap_completed = bool(result.get("full_scan_completed"))
             if bootstrap_completed:
                 _set_full_scan_done(redis_client, True)
+                _clear_sync_checkpoint(redis_client)
                 logger.info("Bootstrap full scan completed; hourly schedule continues")
             else:
                 _set_full_scan_done(redis_client, False)
                 logger.info("Bootstrap full scan not complete yet; queuing immediate continuation")
                 _enqueue_bootstrap_followup("bootstrap_not_completed")
+        elif result.get("status") == "success":
+            _clear_sync_checkpoint(redis_client)
 
         summary = {
             "task_id": task_id,