fix sold cars update

dubizzle_scraper/worker/tasks.py

@@ -63,6 +63,7 @@ SYNC_SEGMENT_LOCK_KEY_FMT = "dubizzle:locks:sync_segment:{idx}"
 SYNC_SEGMENTS_PROGRESS_KEY = "dubizzle:state:sync_segments_progress"
 SYNC_SEGMENTS_TOTAL_KEY = "dubizzle:state:sync_segments_total"
 SYNC_SEGMENTED_SCAN_ACTIVE_KEY = "dubizzle:state:sync_segmented_scan_active"
+SYNC_SEGMENTED_ACTIVE_URLS_KEY = "dubizzle:state:sync_segmented_active_urls"
 SYNC_SEGMENTS_PROGRESS_TTL_SECONDS = 24 * 60 * 60
 TASK_PROGRESS_KEY_FMT = "dubizzle:state:task_progress:{task_id}"
 GLOBAL_PROGRESS_TS_KEY = "dubizzle:state:last_progress_ts"
@@ -89,6 +90,7 @@ SYNC_SEGMENT_LOCK_KEY_FMT = _runtime_key(SYNC_SEGMENT_LOCK_KEY_FMT)
 SYNC_SEGMENTS_PROGRESS_KEY = _runtime_key(SYNC_SEGMENTS_PROGRESS_KEY)
 SYNC_SEGMENTS_TOTAL_KEY = _runtime_key(SYNC_SEGMENTS_TOTAL_KEY)
 SYNC_SEGMENTED_SCAN_ACTIVE_KEY = _runtime_key(SYNC_SEGMENTED_SCAN_ACTIVE_KEY)
+SYNC_SEGMENTED_ACTIVE_URLS_KEY = _runtime_key(SYNC_SEGMENTED_ACTIVE_URLS_KEY)
 TASK_PROGRESS_KEY_FMT = _runtime_key(TASK_PROGRESS_KEY_FMT)
 GLOBAL_PROGRESS_TS_KEY = _runtime_key(GLOBAL_PROGRESS_TS_KEY)
 SITEMAP_HOURLY_LAST_COUNT_KEY = _runtime_key(SITEMAP_HOURLY_LAST_COUNT_KEY)
@@ -736,6 +738,43 @@ def _clear_segmented_scan_active(redis_client: Redis) -> None:
         logger.warning("Failed to clear segmented scan active state", exc_info=True)
 
 
+def _add_segmented_active_urls(redis_client: Redis, urls: list[str] | set[str]) -> int:
+    normalized_urls = [str(url).strip() for url in urls if str(url).strip()]
+    if not normalized_urls:
+        return 0
+
+    try:
+        added = 0
+        pipe = redis_client.pipeline()
+        chunk_size = 5000
+        for start in range(0, len(normalized_urls), chunk_size):
+            chunk = normalized_urls[start:start + chunk_size]
+            pipe.sadd(SYNC_SEGMENTED_ACTIVE_URLS_KEY, *chunk)
+            pipe.expire(SYNC_SEGMENTED_ACTIVE_URLS_KEY, SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
+            results = pipe.execute()
+            added += int(results[0] or 0)
+        return added
+    except Exception:
+        logger.warning("Failed to persist segmented active URLs", exc_info=True)
+        return 0
+
+
+def _load_segmented_active_urls(redis_client: Redis) -> set[str]:
+    try:
+        raw = redis_client.smembers(SYNC_SEGMENTED_ACTIVE_URLS_KEY) or set()
+    except Exception:
+        logger.warning("Failed to read segmented active URLs", exc_info=True)
+        return set()
+    return {str(url).strip() for url in raw if str(url).strip()}
+
+
+def _clear_segmented_active_urls(redis_client: Redis) -> None:
+    try:
+        redis_client.delete(SYNC_SEGMENTED_ACTIVE_URLS_KEY)
+    except Exception:
+        logger.warning("Failed to clear segmented active URLs", exc_info=True)
+
+
 def _is_segmented_scan_in_progress(redis_client: Redis, celery_app) -> bool:
     try:
         marker = redis_client.get(SYNC_SEGMENTED_SCAN_ACTIVE_KEY)
@@ -996,6 +1035,7 @@ def _reset_segments_progress(redis_client: Redis, total: int) -> None:
     try:
         pipe = redis_client.pipeline()
         pipe.delete(SYNC_SEGMENTS_PROGRESS_KEY)
+        pipe.delete(SYNC_SEGMENTED_ACTIVE_URLS_KEY)
         pipe.set(SYNC_SEGMENTS_TOTAL_KEY, str(int(total)), ex=SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
         pipe.execute()
     except Exception:
@@ -1110,6 +1150,12 @@ def sync_segment_task(
 
         result = _run_browser_job(_job)
         segment_done = bool(result.get("full_scan_completed", False))
+        listing_payload = result.get("listing") if isinstance(result.get("listing"), dict) else {}
+        active_urls = listing_payload.get("vehicle_urls") or []
+        active_urls_count = len(active_urls)
+        if active_urls:
+            _add_segmented_active_urls(redis_client, active_urls)
+
         _update_task_progress(
             redis_client,
             task_id=task_id,
@@ -1129,10 +1175,29 @@ def sync_segment_task(
                 segment_index, seg_label, completed, total,
             )
             if total > 0 and completed >= total:
+                sold_count = 0
+                active_urls_for_scan = _load_segmented_active_urls(redis_client)
+                if active_urls_for_scan:
+                    try:
+                        sold_count = persistence.mark_sold_not_in_listing_by_urls(
+                            active_urls_for_scan,
+                            lane=_origin_prefix().rstrip(":"),
+                        )
+                        logger.info(
+                            "Segmented full scan sold-mark completed: sold=%d active_urls=%d",
+                            sold_count,
+                            len(active_urls_for_scan),
+                        )
+                    except Exception:
+                        logger.warning("Segmented full scan sold-mark failed", exc_info=True)
+                else:
+                    logger.warning("Segmented full scan completed without aggregated active URLs; skip sold-mark")
+
                 always_full_scan = bool(Settings().discovery.always_full_scan)
                 _set_full_scan_done(redis_client, False if always_full_scan else True)
                 _clear_sync_checkpoint(redis_client)
                 _clear_segmented_scan_active(redis_client)
+                _clear_segmented_active_urls(redis_client)
                 _clear_bootstrap_failure_streak(redis_client)
                 _clear_bootstrap_continuation_streak(redis_client)
                 if always_full_scan:
@@ -1146,6 +1211,7 @@ def sync_segment_task(
             "segment_label": seg_label,
             "cars_upserted": result.get("cars_upserted", 0),
             "cars_failed": result.get("cars_failed", 0),
+            "active_urls_collected": active_urls_count,
             "vehicles_collected": result.get("listing", {}).get("vehicles_collected", 0),
             "full_scan_completed": segment_done,
         }