fix sold sync tracking

dubizzle_scraper/worker/tasks.py

@@ -64,6 +64,7 @@ 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_SEGMENTED_ACTIVE_IDS_KEY = "dubizzle:state:sync_segmented_active_ids"
 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"
@@ -91,6 +92,7 @@ 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)
+SYNC_SEGMENTED_ACTIVE_IDS_KEY = _runtime_key(SYNC_SEGMENTED_ACTIVE_IDS_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)
@@ -759,6 +761,27 @@ def _add_segmented_active_urls(redis_client: Redis, urls: list[str] | set[str])
         return 0
 
 
+def _add_segmented_active_ids(redis_client: Redis, origin_ids: list[str] | set[str]) -> int:
+    normalized_ids = [str(origin_id).strip() for origin_id in origin_ids if str(origin_id).strip()]
+    if not normalized_ids:
+        return 0
+
+    try:
+        added = 0
+        pipe = redis_client.pipeline()
+        chunk_size = 5000
+        for start in range(0, len(normalized_ids), chunk_size):
+            chunk = normalized_ids[start:start + chunk_size]
+            pipe.sadd(SYNC_SEGMENTED_ACTIVE_IDS_KEY, *chunk)
+            pipe.expire(SYNC_SEGMENTED_ACTIVE_IDS_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 origin_ids", 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()
@@ -768,6 +791,15 @@ def _load_segmented_active_urls(redis_client: Redis) -> set[str]:
     return {str(url).strip() for url in raw if str(url).strip()}
 
 
+def _load_segmented_active_ids(redis_client: Redis) -> set[str]:
+    try:
+        raw = redis_client.smembers(SYNC_SEGMENTED_ACTIVE_IDS_KEY) or set()
+    except Exception:
+        logger.warning("Failed to read segmented active origin_ids", exc_info=True)
+        return set()
+    return {str(origin_id).strip() for origin_id in raw if str(origin_id).strip()}
+
+
 def _clear_segmented_active_urls(redis_client: Redis) -> None:
     try:
         redis_client.delete(SYNC_SEGMENTED_ACTIVE_URLS_KEY)
@@ -775,6 +807,13 @@ def _clear_segmented_active_urls(redis_client: Redis) -> None:
         logger.warning("Failed to clear segmented active URLs", exc_info=True)
 
 
+def _clear_segmented_active_ids(redis_client: Redis) -> None:
+    try:
+        redis_client.delete(SYNC_SEGMENTED_ACTIVE_IDS_KEY)
+    except Exception:
+        logger.warning("Failed to clear segmented active origin_ids", 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)
@@ -1036,6 +1075,7 @@ def _reset_segments_progress(redis_client: Redis, total: int) -> None:
         pipe = redis_client.pipeline()
         pipe.delete(SYNC_SEGMENTS_PROGRESS_KEY)
         pipe.delete(SYNC_SEGMENTED_ACTIVE_URLS_KEY)
+        pipe.delete(SYNC_SEGMENTED_ACTIVE_IDS_KEY)
         pipe.set(SYNC_SEGMENTS_TOTAL_KEY, str(int(total)), ex=SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
         pipe.execute()
     except Exception:
@@ -1152,9 +1192,12 @@ def sync_segment_task(
         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_ids = result.get("all_listing_origin_ids") or []
         active_urls_count = len(active_urls)
         if active_urls:
             _add_segmented_active_urls(redis_client, active_urls)
+        if active_ids:
+            _add_segmented_active_ids(redis_client, active_ids)
 
         _update_task_progress(
             redis_client,
@@ -1176,8 +1219,22 @@ def sync_segment_task(
             )
             if total > 0 and completed >= total:
                 sold_count = 0
+                active_ids_for_scan = _load_segmented_active_ids(redis_client)
                 active_urls_for_scan = _load_segmented_active_urls(redis_client)
-                if active_urls_for_scan:
+                if active_ids_for_scan:
+                    try:
+                        sold_count = persistence.mark_sold_not_in_listing(
+                            active_ids_for_scan,
+                            lane=_origin_prefix().rstrip(":"),
+                        )
+                        logger.info(
+                            "Segmented full scan sold-mark completed by origin_id: sold=%d active_ids=%d",
+                            sold_count,
+                            len(active_ids_for_scan),
+                        )
+                    except Exception:
+                        logger.warning("Segmented full scan sold-mark by origin_id failed", exc_info=True)
+                elif active_urls_for_scan:
                     try:
                         sold_count = persistence.mark_sold_not_in_listing_by_urls(
                             active_urls_for_scan,
@@ -1198,6 +1255,7 @@ def sync_segment_task(
                 _clear_sync_checkpoint(redis_client)
                 _clear_segmented_scan_active(redis_client)
                 _clear_segmented_active_urls(redis_client)
+                _clear_segmented_active_ids(redis_client)
                 _clear_bootstrap_failure_streak(redis_client)
                 _clear_bootstrap_continuation_streak(redis_client)
                 if always_full_scan: