diff --git a/dubizzle_scraper/worker/tasks.py b/dubizzle_scraper/worker/tasks.py index 4c3030f..e724d1d 100644 --- a/dubizzle_scraper/worker/tasks.py +++ b/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: