From 0e8f5bbd92ab1e7d94df3a9fb998053e5dc2a5fa Mon Sep 17 00:00:00 2001 From: qananasikq Date: Mon, 27 Apr 2026 13:06:57 +0300 Subject: [PATCH] fix sold cars update --- dubizzle_scraper/worker/tasks.py | 66 ++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/dubizzle_scraper/worker/tasks.py b/dubizzle_scraper/worker/tasks.py index 239a738..4c3030f 100644 --- a/dubizzle_scraper/worker/tasks.py +++ b/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, }