From 3c0a1da2bf5a3d6fb88ef532139dbc9e87922a99 Mon Sep 17 00:00:00 2001 From: qananasikq Date: Tue, 28 Apr 2026 22:14:13 +0300 Subject: [PATCH] Bootstrap incremental --- iaai_scraper/worker/tasks.py | 1618 ++++++++++++++++++++++++++++------ 1 file changed, 1326 insertions(+), 292 deletions(-) diff --git a/iaai_scraper/worker/tasks.py b/iaai_scraper/worker/tasks.py index 06522ab..be68304 100644 --- a/iaai_scraper/worker/tasks.py +++ b/iaai_scraper/worker/tasks.py @@ -30,17 +30,33 @@ MOBILEDE_SEARCH_CURSOR_KEY = "mobilede:state:search_next_page" MOBILEDE_SEGMENT_CURSOR_KEY_FMT = "mobilede:state:search_next_page:{segment_key}" MOBILEDE_RUNTIME_SEGMENT_INDEX_KEY = "mobilede:state:runtime_segment_index" MOBILEDE_RUNTIME_SEGMENTS_TASK = "mobilede.sync_runtime_segments" +MOBILEDE_SYNC_TASK_NAME = "mobilede.sync_search" +MOBILEDE_SEGMENT_LOCK_KEY_FMT = "mobilede:locks:segment:{segment_key}" +MOBILEDE_SEGMENT_FOLLOWUP_PENDING_KEY_FMT = "mobilede:state:followup_pending:{segment_key}" MOBILEDE_PROGRESS_PAGE_COUNTER_KEY_FMT = "mobilede:state:progress_pages:{segment_key}" -MOBILEDE_CONTINUOUS_SYNC_ENABLED = os.getenv("MOBILEDE_CONTINUOUS_SYNC_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_CONTINUOUS_SYNC_ENABLED = os.getenv("MOBILEDE_CONTINUOUS_SYNC_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS = max(0, int(float(os.getenv("MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS", "15")))) -MOBILEDE_PROGRESS_LOG_EVERY_PAGES = max(1, int(os.getenv("MOBILEDE_PROGRESS_LOG_EVERY_PAGES", "50"))) +MOBILEDE_PROGRESS_LOG_EVERY_PAGES = max(1, int(os.getenv("MOBILEDE_PROGRESS_LOG_EVERY_PAGES", "10"))) MOBILEDE_SKIP_EMPTY_WINDOW = os.getenv("MOBILEDE_SKIP_EMPTY_WINDOW", "true").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_ROTATE_RUNTIME_SEGMENTS = os.getenv("MOBILEDE_ROTATE_RUNTIME_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_RUNTIME_INITIAL_TASKS = max(1, int(os.getenv("MOBILEDE_RUNTIME_INITIAL_TASKS", "2"))) MOBILEDE_SEGMENT_PAGE_WINDOW = max(1, int(os.getenv("MOBILEDE_SEGMENT_PAGE_WINDOW", "10"))) -MOBILEDE_SEGMENT_TARGET_RESULTS = max(100, int(os.getenv("MOBILEDE_SEGMENT_TARGET_RESULTS", "1800"))) +MOBILEDE_RESULTS_PER_PAGE = max(1, int(os.getenv("MOBILEDE_RESULTS_PER_PAGE", "20"))) +MOBILEDE_MAX_PAGE_NUMBER = max(1, int(os.getenv("MOBILEDE_MAX_PAGE_NUMBER", "50"))) +MOBILEDE_SEGMENT_TARGET_RESULTS = max( + MOBILEDE_RESULTS_PER_PAGE, + int(os.getenv("MOBILEDE_SEGMENT_TARGET_RESULTS", str(MOBILEDE_RESULTS_PER_PAGE * MOBILEDE_MAX_PAGE_NUMBER))), +) MOBILEDE_DYNAMIC_SEGMENT_PROBES = os.getenv("MOBILEDE_DYNAMIC_SEGMENT_PROBES", "false").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = os.getenv("MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE", "false").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_SKIP_EMPTY_DYNAMIC_SEGMENTS = os.getenv("MOBILEDE_SKIP_EMPTY_DYNAMIC_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_HOT_BASE_SPLIT_ENABLED = os.getenv("MOBILEDE_HOT_BASE_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_HOT_BASE_PRICE_MAX = max(5000, int(os.getenv("MOBILEDE_HOT_BASE_PRICE_MAX", "30000"))) +MOBILEDE_HOT_RECENT_YEAR_MIN = max(2000, int(os.getenv("MOBILEDE_HOT_RECENT_YEAR_MIN", "2018"))) +MOBILEDE_HOT_MILEAGE_SPLIT_ENABLED = os.getenv("MOBILEDE_HOT_MILEAGE_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_HOT_MILEAGE_PRICE_MIN = max(1, int(os.getenv("MOBILEDE_HOT_MILEAGE_PRICE_MIN", "15001"))) +MOBILEDE_HOT_MILEAGE_PRICE_MAX = max(MOBILEDE_HOT_MILEAGE_PRICE_MIN, int(os.getenv("MOBILEDE_HOT_MILEAGE_PRICE_MAX", "30000"))) +MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX = max(1, int(os.getenv("MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX", "5000"))) MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED = os.getenv("MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP = os.getenv("MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP", "true").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_INCREMENTAL_PAGE_WINDOW = max(1, int(os.getenv("MOBILEDE_INCREMENTAL_PAGE_WINDOW", "1"))) @@ -54,9 +70,31 @@ MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO = min(1.0, max(0.0, float(os.getenv("MOBILEDE MOBILEDE_BOOTSTRAP_DONE_KEY = "mobilede:state:bootstrap_full_scan_done" MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY = "mobilede:state:bootstrap_segments_total" MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY = "mobilede:state:bootstrap_segments_done" +MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY = "mobilede:state:bootstrap_listings_total" +MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY = "mobilede:state:bootstrap_unique_total" +MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY = "mobilede:state:bootstrap_inserted_total" +MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY = "mobilede:state:bootstrap_updated_total" +MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY = "mobilede:state:bootstrap_images_total" +MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY = "mobilede:state:bootstrap_dispatched_segments" +MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY = "mobilede:state:bootstrap_incremental_transition" MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY = "mobilede:state:runtime_segments_cache" MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY = "mobilede:state:runtime_segments_building" MOBILEDE_RUNTIME_SEGMENTS_PENDING_KEY = "mobilede:state:runtime_segments_pending" +MOBILEDE_RUNTIME_SEGMENTS_CACHE_LOCK_KEY = "mobilede:locks:runtime_segments_cache" +MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY = "mobilede:state:overflow_expanded_parents" +MOBILEDE_OVERFLOW_SPLIT_ENABLED = os.getenv("MOBILEDE_OVERFLOW_SPLIT_ENABLED", "true").strip().lower() in {"1", "true", "yes", "on"} +MOBILEDE_OVERFLOW_SPLIT_THRESHOLD_RATIO = min( + 1.0, + max(0.5, float(os.getenv("MOBILEDE_OVERFLOW_SPLIT_THRESHOLD_RATIO", "0.98"))), +) +MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS = max( + 1, + min(5, int(os.getenv("MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS", "3"))), +) +MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH = max( + 1, + min(6, int(os.getenv("MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH", "3"))), +) MOBILEDE_INCREMENTAL_CYCLE_KEY = "mobilede:state:incremental_cycle" MOBILEDE_INCREMENTAL_CYCLE_SEEN_COUNT_KEY = "mobilede:state:incremental_cycle_seen_count" MOBILEDE_INCREMENTAL_CYCLE_SEEN_SET_KEY_FMT = "mobilede:state:incremental_cycle_seen:{cycle_id}" @@ -172,6 +210,14 @@ def _task_progress_key(task_id: str) -> str: return TASK_PROGRESS_KEY_FMT.format(task_id=task_id) +def _mobilede_segment_lock_key(segment_key: str) -> str: + return MOBILEDE_SEGMENT_LOCK_KEY_FMT.format(segment_key=segment_key) + + +def _mobilede_followup_pending_key(segment_key: str) -> str: + return MOBILEDE_SEGMENT_FOLLOWUP_PENDING_KEY_FMT.format(segment_key=segment_key) + + def _update_task_progress( redis_client: Redis, *, @@ -674,6 +720,76 @@ def _mobilede_segment_key(segment: dict[str, object] | None) -> str: return f"{make_id}:{model_id}".replace(" ", "_") +def _mobilede_task_segment_key( + *, + segment: dict[str, object] | None, + search_url: str | None, + make_id: str | None, + model_id: str | None, + price_min: str | None, + price_max: str | None, + year_min: str | None, + year_max: str | None, + mileage_min: str | None, + mileage_max: str | None, +) -> str: + if segment: + return _mobilede_segment_fingerprint(segment) + payload = { + "search_url": str(search_url or "").strip(), + "make_id": str(make_id or "").strip(), + "model_id": str(model_id or "").strip(), + "price_min": str(price_min or "").strip(), + "price_max": str(price_max or "").strip(), + "year_min": str(year_min or "").strip(), + "year_max": str(year_max or "").strip(), + "mileage_min": str(mileage_min or "").strip(), + "mileage_max": str(mileage_max or "").strip(), + } + return hashlib.sha1(json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")).hexdigest()[:16] + + +def _try_set_mobilede_followup_pending(redis_client: Redis, *, segment_key: str, ttl_seconds: int) -> bool: + try: + return bool(redis_client.set(_mobilede_followup_pending_key(segment_key), "1", nx=True, ex=max(60, int(ttl_seconds)))) + except Exception: + logger.warning("Failed to set mobile.de follow-up pending flag", exc_info=True) + return True + + +def _clear_mobilede_followup_pending(redis_client: Redis, *, segment_key: str) -> None: + try: + redis_client.delete(_mobilede_followup_pending_key(segment_key)) + except Exception: + logger.warning("Failed to clear mobile.de follow-up pending flag", exc_info=True) + + +def _mobilede_followup_pending_is_stale(redis_client: Redis, *, segment_key: str) -> bool: + try: + pending_key = _mobilede_followup_pending_key(segment_key) + if not redis_client.exists(pending_key): + return False + segment_lock_key = _mobilede_segment_lock_key(segment_key) + if redis_client.exists(segment_lock_key): + return False + return True + except Exception: + logger.warning("Failed to inspect mobile.de follow-up pending flag", exc_info=True) + return False + + +def _try_reset_stale_mobilede_followup_pending(redis_client: Redis, *, segment_key: str) -> bool: + if not _mobilede_followup_pending_is_stale(redis_client, segment_key=segment_key): + return False + try: + redis_client.delete(_mobilede_followup_pending_key(segment_key)) + logger.warning("Reset stale mobile.de follow-up pending flag for segment=%s", segment_key) + return True + except Exception: + logger.warning("Failed to reset stale mobile.de follow-up pending flag", exc_info=True) + return False + + def _mobilede_segment_fingerprint(segment: dict[str, object] | None) -> str: if not segment: return "all" @@ -775,6 +891,46 @@ def _mobilede_segment_label(segment: dict[str, object] | None) -> str: return label or _mobilede_segment_key(segment) +def _mobilede_short_segment_label(segment: dict[str, object] | None) -> str: + label = _mobilede_segment_label(segment) + if " | " not in label: + return label + parts = [part.strip() for part in label.split(" | ") if part.strip()] + useful_parts = [part for part in parts if part.startswith(("ms=", "price=", "year", "km"))] + return " | ".join(useful_parts) if useful_parts else label + + +def _mobilede_bootstrap_progress(redis_client: Redis) -> tuple[int, int, int]: + done = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY) or 0) + total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or 0) + if total <= 0: + cached_segments = _get_cached_mobilede_runtime_segments(redis_client) or [] + if cached_segments: + total = len(cached_segments) + try: + redis_client.set(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY, str(total)) + except Exception: + logger.debug("Failed to backfill bootstrap total segments", exc_info=True) + left = max(0, total - done) if total > 0 else 0 + return done, total, left + + +def _mobilede_bootstrap_percent(done: int, total: int) -> float: + if total <= 0: + return 0.0 + return min(100.0, max(0.0, (float(done) / float(total)) * 100.0)) + + +def _mobilede_bootstrap_cars_totals(redis_client: Redis) -> tuple[int, int, int, int, int]: + return ( + int(redis_client.get(MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY) or 0), + int(redis_client.get(MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY) or 0), + int(redis_client.get(MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY) or 0), + int(redis_client.get(MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY) or 0), + int(redis_client.get(MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY) or 0), + ) + + def _find_mobilede_runtime_segment( settings: Settings, *, @@ -833,6 +989,10 @@ def _mobilede_filter_source(segment: dict[str, object] | None, search_url: str | def _is_mobilede_transient_request_error(exc: Exception) -> bool: + if isinstance(exc, requests.exceptions.HTTPError): + status_code = getattr(getattr(exc, "response", None), "status_code", None) + if status_code in {408, 409, 425, 429, 500, 502, 503, 504}: + return True if isinstance( exc, ( @@ -910,8 +1070,8 @@ def _log_mobilede_progress_threshold( if previous_total // MOBILEDE_PROGRESS_LOG_EVERY_PAGES == total_pages // MOBILEDE_PROGRESS_LOG_EVERY_PAGES: return logger.info( - "mobile.de progress: segment=%s total_pages=%s delta_pages=%s cars=%s images=%s last_window=%s-%s task_id=%s", - _mobilede_segment_label(segment), + "mobile.de page progress: segment=%s pages_done=%s (+%s) cars_upserted=%s images=%s last_window=%s-%s task_id=%s", + _mobilede_short_segment_label(segment), total_pages, delta_pages, delta_cars, @@ -931,6 +1091,20 @@ def _mobilede_url_query_value(search_url: str, key: str) -> str | None: return None +def _mobilede_url_query_values(search_url: str, key: str) -> list[str]: + values: list[str] = [] + seen: set[str] = set() + for item_key, item_value in parse_qsl(urlsplit(search_url).query, keep_blank_values=True): + if item_key != key: + continue + value = str(item_value or "").strip() + if not value or value in seen: + continue + seen.add(value) + values.append(value) + return values + + def _mobilede_make_segment_url(search_url: str, **params: str | int | None) -> str: return MobileDeClient.build_search_url_from_existing(search_url, page_number=1, **params) @@ -984,6 +1158,53 @@ def _mobilede_try_mark_cycle_segment_seen( return added == 1 +def _mobilede_try_mark_bootstrap_segment_dispatched( + redis_client: Redis, + segment: dict[str, object] | None, + *, + ttl_seconds: int = 24 * 60 * 60, +) -> bool: + fingerprint = _mobilede_segment_fingerprint(segment) + added = int(redis_client.sadd(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY, fingerprint)) + redis_client.expire(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY, ttl_seconds) + return added == 1 + + +def _mobilede_try_recover_stalled_bootstrap_queue( + redis_client: Redis, + *, + queue_name: str = MOBILEDE_SYNC_QUEUE, +) -> bool: + """Сбрасывает залипшие bootstrap-dispatched маркеры, если очередь пуста и нет активного прогресса.""" + if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or _mobilede_bootstrap_done(redis_client): + return False + try: + queue_len = int(redis_client.llen(queue_name) or 0) + if queue_len > 0: + return False + has_active_progress = bool(redis_client.exists("mobilede:state:active_progress")) + if has_active_progress: + return False + done = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY) or 0) + total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or 0) + if total <= 0 or done >= total: + return False + dispatched = int(redis_client.scard(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY) or 0) + if dispatched <= 0: + return False + redis_client.delete(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY) + logger.warning( + "mobile.de bootstrap queue stall recovered: queue=0 active=0 progress=%s/%s dispatched=%s -> cleared", + done, + total, + dispatched, + ) + return True + except Exception: + logger.debug("Failed to recover stalled mobile.de bootstrap queue", exc_info=True) + return False + + def _mobilede_current_cycle_id(redis_client: Redis) -> str: cycle_id = str(redis_client.get(MOBILEDE_INCREMENTAL_CYCLE_KEY) or "").strip() if not cycle_id: @@ -992,6 +1213,23 @@ def _mobilede_current_cycle_id(redis_client: Redis) -> str: return cycle_id +def _mobilede_incremental_cycle_progress( + redis_client: Redis, + *, + cycle_id: str | None, + total_segments: int | None = None, +) -> tuple[str, int, int, int]: + resolved_cycle_id = str(cycle_id or _mobilede_current_cycle_id(redis_client) or "1").strip() or "1" + total = max(0, int(total_segments or 0)) + if total <= 0: + total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or 0) + if total <= 0: + total = len(_get_cached_mobilede_runtime_segments(redis_client) or []) + seen = int(redis_client.scard(_mobilede_cycle_seen_set_key(resolved_cycle_id)) or 0) + left = max(0, total - seen) if total > 0 else 0 + return resolved_cycle_id, seen, total, left + + def _mobilede_reserve_strict_first_pass_segment( redis_client: Redis, settings: Settings, @@ -1095,6 +1333,25 @@ def _mobilede_year_ranges() -> list[tuple[int | None, int | None]]: return [(None, 1999), (2000, 2004), (2005, 2009), (2010, 2014), (2015, 2017), (2018, 2020), (2021, 2022), (2023, 2024), (2025, None)] +def _mobilede_hot_year_ranges() -> list[tuple[int | None, int | None]]: + return [(None, 2009), (2010, 2017), (2018, 2020), (2021, 2022), (2023, 2024), (2025, None)] + + +def _mobilede_low_price_hot_year_ranges() -> list[tuple[int | None, int | None]]: + return [(None, 2004), (2005, 2009), (2010, 2013), (2014, 2017), (2018, 2020), (2021, 2022), (2023, 2024), (2025, None)] + + +def _mobilede_year_ranges_for_price(price_min: int, price_max: int | None) -> list[tuple[int | None, int | None]]: + if not MOBILEDE_HOT_BASE_SPLIT_ENABLED: + return _mobilede_year_ranges() + upper_bound = int(price_max) if price_max is not None else int(price_min) + if upper_bound <= 10000: + return _mobilede_low_price_hot_year_ranges() + if upper_bound <= MOBILEDE_HOT_BASE_PRICE_MAX: + return _mobilede_hot_year_ranges() + return _mobilede_year_ranges() + + def _mobilede_mileage_ranges() -> list[tuple[int | None, int | None]]: compact = os.getenv("MOBILEDE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"} if compact: @@ -1102,6 +1359,95 @@ def _mobilede_mileage_ranges() -> list[tuple[int | None, int | None]]: return [(None, 50000), (50001, 100000), (100001, 150000), (150001, 200000), (200001, None)] +def _mobilede_should_pre_split_mileage( + price_min: int, + price_max: int | None, + year_min: int | None, + year_max: int | None, +) -> bool: + if not MOBILEDE_HOT_MILEAGE_SPLIT_ENABLED: + return False + + # Дешёвый старый сегмент — один из самых плотных для Toyota/Hyundai. + if price_max is not None and price_max <= MOBILEDE_HOT_OLD_CHEAP_PRICE_MAX: + return year_max is not None and year_max <= 2009 + + if ( + price_max is not None + and price_max <= 10000 + and year_min is not None + and year_min >= 2010 + and year_max is not None + and year_max <= 2017 + ): + return True + + if ( + price_min >= 10001 + and price_max is not None + and price_max <= 15000 + and year_min is not None + and year_min >= 2010 + and year_max is not None + and year_max <= 2020 + ): + return True + + if ( + price_min >= 15001 + and price_max is not None + and price_max <= 30000 + and year_min is not None + and year_min >= 2021 + ): + return True + + if ( + price_min >= 30001 + and price_max is not None + and price_max <= 75000 + and year_min is not None + and year_min >= 2023 + ): + return True + + # Самый ликвидный recent-mid-price диапазон сразу режем по пробегу, + # чтобы не терять хвост за лимитом 50x20 и не плодить поздний overflow. + return bool( + year_min is not None + and year_min >= MOBILEDE_HOT_RECENT_YEAR_MIN + and price_min >= MOBILEDE_HOT_MILEAGE_PRICE_MIN + and price_max is not None + and price_max <= MOBILEDE_HOT_MILEAGE_PRICE_MAX + ) + + +def _mobilede_price_subranges_for_hot_year( + price_min: int, + price_max: int | None, + year_min: int | None, + year_max: int | None, +) -> list[tuple[int, int | None]]: + if price_max is None: + return [(price_min, price_max)] + + if year_min is not None and year_min >= 2025: + if price_min == 15001 and price_max == 20000: + return [(15001, 17500), (17501, 20000)] + if price_min == 20001 and price_max == 30000: + return [(20001, 25000), (25001, 30000)] + + if year_min is not None and year_min >= 2023: + if price_min == 20001 and price_max == 30000: + return [(20001, 25000), (25001, 30000)] + if price_min == 30001 and price_max == 50000: + return [(30001, 40000), (40001, 50000)] + if price_min == 50001 and price_max == 75000: + return [(50001, 62500), (62501, 75000)] + + return [(price_min, price_max)] + + def _mobilede_range_value(min_value: int | None, max_value: int | None) -> str: return f"{min_value or ''}:{max_value or ''}" @@ -1126,6 +1472,304 @@ def _mobilede_mileage_label(mileage_min: int | None, mileage_max: int | None) -> return f"km={mileage_min}-{mileage_max}" +def _mobilede_overflow_threshold(max_pages: int) -> int: + cap = max(MOBILEDE_RESULTS_PER_PAGE, int(max_pages) * MOBILEDE_RESULTS_PER_PAGE) + threshold = int(float(cap) * MOBILEDE_OVERFLOW_SPLIT_THRESHOLD_RATIO) + return max(MOBILEDE_RESULTS_PER_PAGE, min(cap, threshold)) + + +def _mobilede_parse_optional_int(value: object) -> int | None: + raw = str(value or "").strip() + if not raw: + return None + try: + return int(raw) + except ValueError: + return None + + +def _mobilede_split_year_ranges_for_overflow( + year_min: int | None, + year_max: int | None, +) -> list[tuple[int | None, int | None]]: + current_year = int(time.gmtime().tm_year) + if year_min is None and year_max is None: + return [] + if year_min is None and year_max is not None: + pivot = int(year_max) - 5 + if pivot <= 0 or pivot >= int(year_max): + return [] + return [(None, pivot), (pivot + 1, int(year_max))] + if year_min is not None and year_max is None: + if int(year_min) >= current_year - 1: + return [] + pivot = min(current_year - 2, int(year_min) + 2) + if pivot <= int(year_min): + return [] + return [(int(year_min), pivot), (pivot + 1, None)] + + # Оба значения заданы. + assert year_min is not None and year_max is not None + span = int(year_max) - int(year_min) + if span < 2: + return [] + pivot = int(year_min) + span // 2 + if pivot <= int(year_min) or pivot >= int(year_max): + return [] + return [(int(year_min), pivot), (pivot + 1, int(year_max))] + + +def _mobilede_split_price_ranges_for_overflow( + price_min: int | None, + price_max: int | None, +) -> list[tuple[int, int | None]]: + left = int(price_min or 1) + right = price_max + if right is None: + step = max(5000, min(50000, left)) + pivot = left + step + return [(left, pivot), (pivot + 1, None)] + span = int(right) - int(left) + if span < 2000: + return [] + pivot = int(left) + span // 2 + if pivot <= int(left) or pivot >= int(right): + return [] + return [(int(left), pivot), (pivot + 1, int(right))] + + +def _mobilede_make_overflow_child_segment( + segment: dict[str, object], + *, + search_url: str, + max_pages: int, + parent_fingerprint: str, + depth: int, + split_kind: str, + split_label: str, + split_params: dict[str, str], + price_range: tuple[int | None, int | None] | None = None, + year_range: tuple[int | None, int | None] | None = None, + mileage_range: tuple[int | None, int | None] | None = None, +) -> dict[str, object]: + child = dict(segment) + child["search_url"] = _mobilede_make_segment_url(search_url, **split_params) + child["listing_url"] = child["search_url"] + child["start_page"] = 1 + child["max_pages"] = max_pages + child["total_results"] = None + child["overflow_parent"] = parent_fingerprint + child["overflow_depth"] = depth + 1 + child["overflow_split"] = split_kind + child["label"] = f"{str(segment.get('label') or 'mobile.de overflow')} | {split_label} | overflow:d{depth + 1}" + + if price_range is not None: + price_min, price_max = price_range + child["price_min"] = str(price_min) if price_min is not None else None + child["price_max"] = str(price_max) if price_max is not None else None + if year_range is not None: + year_min, year_max = year_range + child["year_min"] = str(year_min) if year_min is not None else None + child["year_max"] = str(year_max) if year_max is not None else None + if mileage_range is not None: + mileage_min, mileage_max = mileage_range + child["mileage_min"] = str(mileage_min) if mileage_min is not None else None + child["mileage_max"] = str(mileage_max) if mileage_max is not None else None + return child + + +def _mobilede_build_overflow_child_segments( + *, + segment: dict[str, object], + max_pages: int, +) -> list[dict[str, object]]: + if MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS <= 0: + return [] + + search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() + if not search_url: + return [] + + depth = max(0, int(_mobilede_parse_optional_int(segment.get("overflow_depth")) or 0)) + if depth >= MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH: + return [] + + query_keys = {key for key, _ in parse_qsl(urlsplit(search_url).query, keep_blank_values=True)} + parent_fingerprint = _mobilede_segment_fingerprint(segment) + segment_max_pages = max(1, int(max_pages or segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER)) + + has_mileage_filter = bool( + str(segment.get("mileage_min") or "").strip() + or str(segment.get("mileage_max") or "").strip() + or "ml" in query_keys + ) + + # 1) Первый уровень: mileage-разбиение (самый дешёвый и наименее дублящийся). + if not has_mileage_filter: + child_segments: list[dict[str, object]] = [] + for mileage_min, mileage_max in _mobilede_mileage_ranges()[:MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS]: + child_segments.append( + _mobilede_make_overflow_child_segment( + segment, + search_url=search_url, + max_pages=segment_max_pages, + parent_fingerprint=parent_fingerprint, + depth=depth, + split_kind="mileage", + split_label=_mobilede_mileage_label(mileage_min, mileage_max), + split_params={"ml": _mobilede_range_value(mileage_min, mileage_max)}, + mileage_range=(mileage_min, mileage_max), + ) + ) + return child_segments + + # 2) Если mileage уже есть — делим год. + year_min = _mobilede_parse_optional_int(segment.get("year_min")) + year_max = _mobilede_parse_optional_int(segment.get("year_max")) + year_splits = _mobilede_split_year_ranges_for_overflow(year_min, year_max) + if year_splits: + child_segments = [] + for split_year_min, split_year_max in year_splits[:2]: + child_segments.append( + _mobilede_make_overflow_child_segment( + segment, + search_url=search_url, + max_pages=segment_max_pages, + parent_fingerprint=parent_fingerprint, + depth=depth, + split_kind="year", + split_label=_mobilede_year_label(split_year_min, split_year_max), + split_params={"fr": _mobilede_range_value(split_year_min, split_year_max)}, + year_range=(split_year_min, split_year_max), + ) + ) + return child_segments + + # 3) Fallback: если год уже очень узкий — делим цену. + price_min = _mobilede_parse_optional_int(segment.get("price_min")) + price_max = _mobilede_parse_optional_int(segment.get("price_max")) + price_splits = _mobilede_split_price_ranges_for_overflow(price_min, price_max) + if price_splits: + child_segments = [] + for split_price_min, split_price_max in price_splits[:2]: + price_label = _mobilede_price_label(split_price_min, split_price_max) + child_segments.append( + _mobilede_make_overflow_child_segment( + segment, + search_url=search_url, + max_pages=segment_max_pages, + parent_fingerprint=parent_fingerprint, + depth=depth, + split_kind="price", + split_label=price_label, + split_params={"p": _mobilede_range_value(split_price_min, split_price_max)}, + price_range=(split_price_min, split_price_max), + ) + ) + return child_segments + + return [] + + +def _mobilede_try_expand_overflow_segment( + redis_client: Redis, + settings: Settings, + *, + segment: dict[str, object] | None, + listing_count: int, + unique_count: int, + max_pages: int, + segment_end_page: int, +) -> int: + if not MOBILEDE_OVERFLOW_SPLIT_ENABLED or not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or not segment: + return 0 + + normalized_max_pages = max(1, int(max_pages or segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER)) + if int(segment_end_page) < normalized_max_pages: + return 0 + + observed = max(int(listing_count or 0), int(unique_count or 0)) + threshold = _mobilede_overflow_threshold(normalized_max_pages) + if observed < threshold: + return 0 + + child_segments = _mobilede_build_overflow_child_segments( + segment=segment, + max_pages=normalized_max_pages, + ) + if not child_segments: + return 0 + + parent_fingerprint = _mobilede_segment_fingerprint(segment) + if int(redis_client.sadd(MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY, parent_fingerprint)) != 1: + return 0 + redis_client.expire(MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY, 30 * 24 * 60 * 60) + + lock_owner = f"overflow:{uuid.uuid4().hex}" + if not _acquire_lock(redis_client, MOBILEDE_RUNTIME_SEGMENTS_CACHE_LOCK_KEY, lock_owner, 120): + redis_client.srem(MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY, parent_fingerprint) + return 0 + + committed = False + try: + cached_segments = _get_cached_mobilede_runtime_segments(redis_client) + if cached_segments is None: + cached_segments = _build_mobilede_runtime_segments(settings) + + existing_fingerprints = { + _mobilede_segment_fingerprint(item) + for item in cached_segments + if isinstance(item, dict) + } + appended_segments: list[dict[str, object]] = [] + for child in child_segments: + child_fingerprint = _mobilede_segment_fingerprint(child) + if child_fingerprint in existing_fingerprints: + continue + existing_fingerprints.add(child_fingerprint) + appended_segments.append(child) + + if not appended_segments: + committed = True + return 0 + + updated_segments = [dict(item) for item in cached_segments if isinstance(item, dict)] + appended_segments + redis_client.set( + MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY, + json.dumps(updated_segments, ensure_ascii=False), + ex=24 * 60 * 60, + ) + redis_client.set(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY, str(len(updated_segments))) + done_segments = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY) or 0) + if done_segments < len(updated_segments): + redis_client.delete(MOBILEDE_BOOTSTRAP_DONE_KEY) + + committed = True + logger.info( + "mobile.de overflow split appended: parent=%s observed=%s threshold=%s added=%s total_segments=%s", + _mobilede_short_segment_label(segment), + observed, + threshold, + len(appended_segments), + len(updated_segments), + ) + return len(appended_segments) + except Exception: + logger.warning( + "Failed to append mobile.de overflow segments: parent=%s", + _mobilede_segment_label(segment), + exc_info=True, + ) + return 0 + finally: + if not committed: + try: + redis_client.srem(MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY, parent_fingerprint) + except Exception: + logger.debug("Failed to rollback overflow parent marker", exc_info=True) + _release_lock_if_owner(redis_client, MOBILEDE_RUNTIME_SEGMENTS_CACHE_LOCK_KEY, lock_owner) + + def _mobilede_probe_total(search_url: str, **params: str | int | None) -> int | None: try: client = MobileDeClient.for_worker(delay_seconds=0) @@ -1138,8 +1782,8 @@ def _mobilede_probe_total(search_url: str, **params: str | int | None) -> int | def _mobilede_segment_pages_for_total(total_results: int | None, fallback_max_pages: int) -> int: if total_results is None or total_results <= 0: - return min(fallback_max_pages, 100) - pages = max(1, min(100, (int(total_results) + 19) // 20)) + return min(fallback_max_pages, MOBILEDE_MAX_PAGE_NUMBER) + pages = max(1, min(MOBILEDE_MAX_PAGE_NUMBER, (int(total_results) + MOBILEDE_RESULTS_PER_PAGE - 1) // MOBILEDE_RESULTS_PER_PAGE)) return min(fallback_max_pages, pages) @@ -1163,10 +1807,66 @@ def _mobilede_make_expanded_segment( return item +def _mobilede_set_segment_range_fields( + item: dict[str, object], + *, + price_min: int, + price_max: int | None, + year_range: tuple[int | None, int | None] | None = None, + mileage_range: tuple[int | None, int | None] | None = None, +) -> None: + item["price_min"] = str(price_min) + item["price_max"] = str(price_max) if price_max is not None else None + if year_range is not None: + year_min, year_max = year_range + item["year_min"] = str(year_min) if year_min is not None else None + item["year_max"] = str(year_max) if year_max is not None else None + if mileage_range is not None: + mileage_min, mileage_max = mileage_range + item["mileage_min"] = str(mileage_min) if mileage_min is not None else None + item["mileage_max"] = str(mileage_max) if mileage_max is not None else None + + +def _mobilede_split_search_url_segment_by_make(segment: dict[str, object]) -> list[dict[str, object]]: + search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() + if not search_url: + return [segment] + + make_tokens = _mobilede_url_query_values(search_url, "ms") + if len(make_tokens) <= 1: + return [segment] + + base_label = str(segment.get("label") or "mobile.de filtered URL").strip() or "mobile.de filtered URL" + split_segments: list[dict[str, object]] = [] + for make_token in make_tokens: + item = dict(segment) + item["search_url"] = _mobilede_make_segment_url(search_url, ms=make_token) + item["listing_url"] = item["search_url"] + item["start_page"] = int(segment.get("start_page") or 1) + item["make_id"] = make_token + item["label"] = f"{base_label} | ms={make_token}" + split_segments.append(item) + + logger.info( + "mobile.de split multi-make search URL into %s make segment(s): %s", + len(split_segments), + ", ".join(str(item.get("label") or "") for item in split_segments), + ) + return split_segments + + def _expand_mobilede_search_url_segment(segment: dict[str, object]) -> list[dict[str, object]]: search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() if not search_url: return [segment] + + split_by_make = _mobilede_split_search_url_segment_by_make(segment) + if len(split_by_make) > 1: + expanded: list[dict[str, object]] = [] + for split_segment in split_by_make: + expanded.extend(_expand_mobilede_search_url_segment(split_segment)) + return expanded + # Если пользователь уже задал узкий price/year/mileage range — не размножаем автоматически. existing_range_keys = {"p", "fr", "ml"} query_keys = {key for key, _ in parse_qsl(urlsplit(search_url).query, keep_blank_values=True)} @@ -1175,7 +1875,7 @@ def _expand_mobilede_search_url_segment(segment: dict[str, object]) -> list[dict expanded: list[dict[str, object]] = [] base_label = str(segment.get("label") or "mobile.de segmented URL") - max_pages = int(segment.get("max_pages") or 100) + max_pages = int(segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER) for price_min, price_max in _mobilede_price_ranges(): params: dict[str, str | int | None] = {} if price_max is not None: @@ -1196,58 +1896,87 @@ def _expand_mobilede_search_url_segment(segment: dict[str, object]) -> list[dict total_results=price_total, fallback_max_pages=max_pages, ) - item["price_min"] = str(price_min) - item["price_max"] = str(price_max) if price_max is not None else None + _mobilede_set_segment_range_fields(item, price_min=price_min, price_max=price_max) expanded.append(item) continue - for year_min, year_max in _mobilede_year_ranges(): - year_params = dict(params) - year_params["fr"] = _mobilede_range_value(year_min, year_max) + for year_min, year_max in _mobilede_year_ranges_for_price(price_min, price_max): year_label = _mobilede_year_label(year_min, year_max) - year_total = _mobilede_probe_total(search_url, **year_params) if MOBILEDE_DYNAMIC_SEGMENT_PROBES else None - if _mobilede_should_skip_dynamic_segment(year_total): - continue - if year_total is not None and year_total <= MOBILEDE_SEGMENT_TARGET_RESULTS: - item = _mobilede_make_expanded_segment( - segment, - search_url=search_url, - base_label=base_label, - label_parts=[price_label, year_label], - params=year_params, - total_results=year_total, - fallback_max_pages=max_pages, - ) - item["price_min"] = str(price_min) - item["price_max"] = str(price_max) if price_max is not None else None - item["year_min"] = str(year_min) if year_min is not None else None - item["year_max"] = str(year_max) if year_max is not None else None - expanded.append(item) - continue - - for mileage_min, mileage_max in _mobilede_mileage_ranges(): - mileage_params = dict(year_params) - mileage_params["ml"] = _mobilede_range_value(mileage_min, mileage_max) - mileage_label = _mobilede_mileage_label(mileage_min, mileage_max) - mileage_total = _mobilede_probe_total(search_url, **mileage_params) if MOBILEDE_DYNAMIC_SEGMENT_PROBES else None - if _mobilede_should_skip_dynamic_segment(mileage_total): + refined_price_ranges = _mobilede_price_subranges_for_hot_year(price_min, price_max, year_min, year_max) + for refined_price_min, refined_price_max in refined_price_ranges: + refined_price_label = _mobilede_price_label(refined_price_min, refined_price_max) + year_params = dict(params) + year_params["p"] = f"{refined_price_min}:{refined_price_max or ''}" + year_params["fr"] = _mobilede_range_value(year_min, year_max) + year_total = _mobilede_probe_total(search_url, **year_params) if MOBILEDE_DYNAMIC_SEGMENT_PROBES else None + if _mobilede_should_skip_dynamic_segment(year_total): continue - item = _mobilede_make_expanded_segment( - segment, - search_url=search_url, - base_label=base_label, - label_parts=[price_label, year_label, mileage_label], - params=mileage_params, - total_results=mileage_total, - fallback_max_pages=max_pages, - ) - item["price_min"] = str(price_min) - item["price_max"] = str(price_max) if price_max is not None else None - item["year_min"] = str(year_min) if year_min is not None else None - item["year_max"] = str(year_max) if year_max is not None else None - item["mileage_min"] = str(mileage_min) if mileage_min is not None else None - item["mileage_max"] = str(mileage_max) if mileage_max is not None else None - expanded.append(item) + if year_total is not None and year_total <= MOBILEDE_SEGMENT_TARGET_RESULTS: + item = _mobilede_make_expanded_segment( + segment, + search_url=search_url, + base_label=base_label, + label_parts=[refined_price_label, year_label], + params=year_params, + total_results=year_total, + fallback_max_pages=max_pages, + ) + _mobilede_set_segment_range_fields( + item, + price_min=refined_price_min, + price_max=refined_price_max, + year_range=(year_min, year_max), + ) + expanded.append(item) + continue + + if ( + not MOBILEDE_DYNAMIC_SEGMENT_PROBES + and not MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE + and not _mobilede_should_pre_split_mileage(refined_price_min, refined_price_max, year_min, year_max) + ): + item = _mobilede_make_expanded_segment( + segment, + search_url=search_url, + base_label=base_label, + label_parts=[refined_price_label, year_label], + params=year_params, + total_results=year_total, + fallback_max_pages=max_pages, + ) + _mobilede_set_segment_range_fields( + item, + price_min=refined_price_min, + price_max=refined_price_max, + year_range=(year_min, year_max), + ) + expanded.append(item) + continue + + for mileage_min, mileage_max in _mobilede_mileage_ranges(): + mileage_params = dict(year_params) + mileage_params["ml"] = _mobilede_range_value(mileage_min, mileage_max) + mileage_label = _mobilede_mileage_label(mileage_min, mileage_max) + mileage_total = _mobilede_probe_total(search_url, **mileage_params) if MOBILEDE_DYNAMIC_SEGMENT_PROBES else None + if _mobilede_should_skip_dynamic_segment(mileage_total): + continue + item = _mobilede_make_expanded_segment( + segment, + search_url=search_url, + base_label=base_label, + label_parts=[refined_price_label, year_label, mileage_label], + params=mileage_params, + total_results=mileage_total, + fallback_max_pages=max_pages, + ) + _mobilede_set_segment_range_fields( + item, + price_min=refined_price_min, + price_max=refined_price_max, + year_range=(year_min, year_max), + mileage_range=(mileage_min, mileage_max), + ) + expanded.append(item) return expanded @@ -1261,28 +1990,10 @@ def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, objec continue expanded.extend(_expand_mobilede_search_url_segment(segment)) if len(expanded) != len(segments): - logger.info("mobile.de runtime segments expanded: %s -> %s", len(segments), len(expanded)) + logger.info("mobile.de segments planned: input=%s total=%s", len(segments), len(expanded)) return expanded -def _get_mobilede_runtime_segments(redis_client: Redis, settings: Settings) -> list[dict[str, object]]: - try: - cached_raw = redis_client.get(MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY) - if cached_raw: - cached = json.loads(cached_raw) - if isinstance(cached, list): - return [dict(item) for item in cached if isinstance(item, dict)] - except Exception: - logger.debug("Failed to read mobile.de runtime segment cache", exc_info=True) - - segments = _build_mobilede_runtime_segments(settings) - try: - redis_client.set(MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY, json.dumps(segments, ensure_ascii=False), ex=24 * 60 * 60) - except Exception: - logger.debug("Failed to write mobile.de runtime segment cache", exc_info=True) - return segments - - def _get_cached_mobilede_runtime_segments(redis_client: Redis) -> list[dict[str, object]] | None: try: cached_raw = redis_client.get(MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY) @@ -1296,6 +2007,16 @@ def _get_cached_mobilede_runtime_segments(redis_client: Redis) -> list[dict[str, return None +def _mobilede_load_segments_for_reservation(redis_client: Redis, settings: Settings) -> list[dict[str, object]]: + segments = _get_cached_mobilede_runtime_segments(redis_client) + if segments: + return segments + _request_mobilede_runtime_segments_rebuild(redis_client) + if MOBILEDE_DYNAMIC_SEGMENT_PROBES: + return [] + return _build_mobilede_runtime_segments(settings) + + def _request_mobilede_runtime_segments_rebuild(redis_client: Redis) -> None: try: redis_client.set(MOBILEDE_RUNTIME_SEGMENTS_PENDING_KEY, "1", ex=15 * 60) @@ -1303,17 +2024,70 @@ def _request_mobilede_runtime_segments_rebuild(redis_client: Redis) -> None: logger.debug("Failed to request mobile.de runtime segment rebuild", exc_info=True) +def _try_queue_mobilede_incremental_transition( + redis_client: Redis, + *, + lane: str, + delay_seconds: float, + use_cursor: bool, + ttl_seconds: int = 15 * 60, +) -> bool: + try: + if not redis_client.set(MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY, "1", nx=True, ex=max(60, int(ttl_seconds))): + return False + mobilede_sync_runtime_segments_task.apply_async( + kwargs={ + "lane": lane, + "delay_seconds": delay_seconds, + "use_cursor": use_cursor, + "continuous": True, + }, + queue=MOBILEDE_SYNC_QUEUE, + countdown=max(0, int(MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS)), + ) + return True + except Exception: + logger.debug("Failed to queue mobile.de bootstrap->incremental transition", exc_info=True) + try: + redis_client.delete(MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY) + except Exception: + logger.debug("Failed to clear mobile.de bootstrap->incremental transition marker", exc_info=True) + return False + + def _mobilede_bootstrap_done(redis_client: Redis) -> bool: if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED: return True + done = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY) or 0) + total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or 0) + if total > 0: + is_done = done >= total + if not is_done and redis_client.get(MOBILEDE_BOOTSTRAP_DONE_KEY): + redis_client.delete(MOBILEDE_BOOTSTRAP_DONE_KEY) + if not is_done and redis_client.get(MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY): + redis_client.delete(MOBILEDE_BOOTSTRAP_INCREMENTAL_TRANSITION_KEY) + return is_done return bool(redis_client.get(MOBILEDE_BOOTSTRAP_DONE_KEY)) +def _mobilede_force_full_scan_only_new( + only_new: bool | None, + *, + redis_client: Redis | None = None, +) -> bool | None: + """Принудительный full-pass включён только до завершения bootstrap.""" + if only_new is True and redis_client is not None and _mobilede_bootstrap_done(redis_client): + return True + if only_new is True: + return False + return only_new + + def _mobilede_segment_scan_complete(redis_client: Redis, segment: dict[str, object] | None) -> bool: if not segment: return False cursor_raw = redis_client.get(_mobilede_cursor_key(segment)) - segment_max_pages = int(segment.get("max_pages") or 100) + segment_max_pages = int(segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER) return cursor_raw is not None and int(cursor_raw) >= segment_max_pages @@ -1323,9 +2097,19 @@ def _mobilede_bootstrap_segment_done(redis_client: Redis, segment: dict[str, obj return bool(redis_client.get(f"mobilede:state:bootstrap_segment_done:{_mobilede_segment_fingerprint(segment)}")) -def _mark_mobilede_bootstrap_segment_done(redis_client: Redis, segment: dict[str, object] | None, total_segments: int | None) -> None: +def _mark_mobilede_bootstrap_segment_done( + redis_client: Redis, + segment: dict[str, object] | None, + total_segments: int | None, + *, + listings: int = 0, + unique: int = 0, + inserted: int = 0, + updated: int = 0, + images: int = 0, +) -> tuple[int, int, int, int, int, int, int] | None: if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or not segment: - return + return None segment_done_key = f"mobilede:state:bootstrap_segment_done:{_mobilede_segment_fingerprint(segment)}" try: if total_segments is not None: @@ -1334,17 +2118,47 @@ def _mark_mobilede_bootstrap_segment_done(redis_client: Redis, segment: dict[str redis_client.expire(segment_done_key, 30 * 24 * 60 * 60) done = int(redis_client.incr(MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY)) total = int(redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) or total_segments or 0) + total_listings = int(redis_client.incrby(MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY, max(0, int(listings)))) + total_unique = int(redis_client.incrby(MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY, max(0, int(unique)))) + total_inserted = int(redis_client.incrby(MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY, max(0, int(inserted)))) + total_updated = int(redis_client.incrby(MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY, max(0, int(updated)))) + total_images = int(redis_client.incrby(MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY, max(0, int(images)))) + left = max(0, total - done) if total > 0 else 0 + percent = _mobilede_bootstrap_percent(done, total) logger.info( - "mobile.de bootstrap segment completed: %s done=%s/%s", - _mobilede_segment_label(segment), + "mobile.de progress: segment %s/%s done (left=%s, %.1f%%) | segment_cars: listings=%s unique=%s inserted=%s updated=%s images=%s | total_cars: listings=%s unique=%s inserted=%s updated=%s images=%s | segment=%s", done, total, + left, + percent, + int(listings), + int(unique), + int(inserted), + int(updated), + int(images), + total_listings, + total_unique, + total_inserted, + total_updated, + total_images, + _mobilede_short_segment_label(segment), ) if total > 0 and done >= total: redis_client.set(MOBILEDE_BOOTSTRAP_DONE_KEY, "1") - logger.info("mobile.de bootstrap full scan completed: segments=%s", total) + logger.info( + "mobile.de bootstrap full scan completed: segments=%s/%s total_cars: listings=%s unique=%s inserted=%s updated=%s images=%s", + done, + total, + total_listings, + total_unique, + total_inserted, + total_updated, + total_images, + ) + return done, total, left, total_listings, total_unique, total_inserted, total_updated except Exception: logger.debug("Failed to mark mobile.de bootstrap segment complete", exc_info=True) + return None def _reserve_next_mobilede_runtime_segment( @@ -1353,43 +2167,11 @@ def _reserve_next_mobilede_runtime_segment( *, only_new: bool | None = None, ) -> tuple[int, dict[str, object]] | None: - segments = _get_cached_mobilede_runtime_segments(redis_client) - if not segments: - _request_mobilede_runtime_segments_rebuild(redis_client) - segments = _build_mobilede_runtime_segments(settings) if not MOBILEDE_DYNAMIC_SEGMENT_PROBES else [] - if not segments: + reserved = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new) + if reserved is None: return None - hot_only_active = bool(MOBILEDE_ONLY_NEW_HOT_ONLY and only_new is True) - has_hot_segments = _mobilede_has_hot_segments(redis_client, segments) if hot_only_active else False - skip_completed_bootstrap = MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED and not _mobilede_bootstrap_done(redis_client) - - def _try_reserve(*, require_hot: bool, respect_cooldown: bool) -> tuple[int, dict[str, object]] | None: - for _ in range(len(segments)): - next_index = int(redis_client.incr(MOBILEDE_RUNTIME_SEGMENT_INDEX_KEY)) - 1 - segment_index = next_index % len(segments) - segment = segments[segment_index] - if skip_completed_bootstrap and _mobilede_bootstrap_segment_done(redis_client, segment): - continue - if respect_cooldown and _mobilede_segment_in_cooldown(redis_client, segment): - continue - if require_hot and hot_only_active and has_hot_segments and not _mobilede_segment_is_hot(redis_client, segment): - continue - return segment_index + 1, segment - return None - - # 1) Обычный hot-only проход. - reserved = _try_reserve(require_hot=True, respect_cooldown=True) - if reserved is not None: - return reserved - # 2) Если hot-only выжег очередь, разрешаем любые не-cooldown сегменты. - reserved = _try_reserve(require_hot=False, respect_cooldown=True) - if reserved is not None: - return reserved - # 3) Failsafe: если всё в cooldown, берём любой сегмент (иначе парсер встанет). - reserved = _try_reserve(require_hot=False, respect_cooldown=False) - if reserved is not None: - return reserved - return None + segment_index, segment = reserved + return segment_index + 1, segment def _enqueue_mobilede_runtime_segments( @@ -1401,12 +2183,13 @@ def _enqueue_mobilede_runtime_segments( ) -> list[dict[str, object]]: settings = Settings() runtime_config = RuntimeConfig.from_file(settings.runtime_config_file) - only_new = runtime_config.sync.only_new redis_client = _get_redis() - segments = _get_cached_mobilede_runtime_segments(redis_client) - if not segments: - _request_mobilede_runtime_segments_rebuild(redis_client) - segments = _build_mobilede_runtime_segments(settings) if not MOBILEDE_DYNAMIC_SEGMENT_PROBES else [] + only_new = _mobilede_force_full_scan_only_new(runtime_config.sync.only_new, redis_client=redis_client) + full_pass_mode = only_new is not True + effective_use_cursor = use_cursor if only_new is True else False + if runtime_config.sync.only_new is True and only_new is False: + logger.info("mobile.de full-pass mode: forcing only_new=False") + segments = _mobilede_load_segments_for_reservation(redis_client, settings) if not segments: return [] try: @@ -1414,7 +2197,14 @@ def _enqueue_mobilede_runtime_segments( except Exception: logger.debug("Failed to initialize mobile.de bootstrap segment total", exc_info=True) initial_reservations: list[tuple[int, dict[str, object]]] = [] - for _ in range(min(MOBILEDE_RUNTIME_INITIAL_TASKS, len(segments))): + dispatch_count = len(segments) if full_pass_mode else min(MOBILEDE_RUNTIME_INITIAL_TASKS, len(segments)) + if full_pass_mode: + logger.info( + "mobile.de full-pass dispatch: queueing all segments=%s use_cursor=%s", + len(segments), + effective_use_cursor, + ) + for _ in range(dispatch_count): reservation = _reserve_next_mobilede_runtime_segment(redis_client, settings, only_new=only_new) if reservation is None: break @@ -1426,7 +2216,7 @@ def _enqueue_mobilede_runtime_segments( "max_pages": int(segment.get("max_pages") or 5), "lane": lane, "delay_seconds": delay_seconds, - "use_cursor": use_cursor, + "use_cursor": effective_use_cursor, "continuous": continuous, "segment": segment, "segment_index": index, @@ -1443,15 +2233,13 @@ def _reserve_mobilede_runtime_segment( *, only_new: bool | None = None, ) -> tuple[int, dict[str, object]] | None: - segments = _get_cached_mobilede_runtime_segments(redis_client) - if not segments: - _request_mobilede_runtime_segments_rebuild(redis_client) - segments = _build_mobilede_runtime_segments(settings) if not MOBILEDE_DYNAMIC_SEGMENT_PROBES else [] + segments = _mobilede_load_segments_for_reservation(redis_client, settings) if not segments: return None hot_only_active = bool(MOBILEDE_ONLY_NEW_HOT_ONLY and only_new is True) has_hot_segments = _mobilede_has_hot_segments(redis_client, segments) if hot_only_active else False skip_completed_bootstrap = MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED and not _mobilede_bootstrap_done(redis_client) + bootstrap_dispatch_dedupe = skip_completed_bootstrap and only_new is not True def _try_reserve(*, require_hot: bool, respect_cooldown: bool) -> tuple[int, dict[str, object]] | None: for _ in range(len(segments)): @@ -1460,6 +2248,8 @@ def _reserve_mobilede_runtime_segment( segment = segments[segment_index] if skip_completed_bootstrap and _mobilede_bootstrap_segment_done(redis_client, segment): continue + if bootstrap_dispatch_dedupe and not _mobilede_try_mark_bootstrap_segment_dispatched(redis_client, segment): + continue if respect_cooldown and _mobilede_segment_in_cooldown(redis_client, segment): continue if require_hot and hot_only_active and has_hot_segments and not _mobilede_segment_is_hot(redis_client, segment): @@ -2164,7 +2954,7 @@ def mobilede_sync_runtime_segments_task( lane: str = "mobile_de_cars", delay_seconds: float = 0.7, use_cursor: bool = True, - continuous: bool = True, + continuous: bool = False, ): redis_client = _get_redis() owner_token = self.request.id or uuid.uuid4().hex @@ -2172,12 +2962,16 @@ def mobilede_sync_runtime_segments_task( logger.info("mobile.de runtime segment rebuild already running") return {"status": "building"} try: + _mobilede_try_recover_stalled_bootstrap_queue(redis_client) + settings = Settings() cached_segments = _get_cached_mobilede_runtime_segments(redis_client) + runtime_config = RuntimeConfig.from_file(settings.runtime_config_file) + full_pass_mode = _mobilede_force_full_scan_only_new(runtime_config.sync.only_new, redis_client=redis_client) is not True if not cached_segments: - settings = Settings() cached_segments = _build_mobilede_runtime_segments(settings) redis_client.set(MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY, json.dumps(cached_segments, ensure_ascii=False), ex=24 * 60 * 60) redis_client.set(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY, str(len(cached_segments))) + redis_client.delete(MOBILEDE_OVERFLOW_EXPANDED_PARENTS_KEY) redis_client.delete(MOBILEDE_RUNTIME_SEGMENTS_PENDING_KEY) logger.info("mobile.de runtime segments rebuilt: %s", len(cached_segments)) segments = _enqueue_mobilede_runtime_segments( @@ -2199,9 +2993,12 @@ def mobilede_sync_runtime_segments_task( ) return {"status": "fallback", "segments": 0} logger.info( - "mobilede_sync_runtime_segments_task queued %s runtime segment(s): %s", + "mobile.de queue started: total_segments=%s queued_now=%s remaining_after_initial=%s mode=%s first_segments=%s", len(segments), - ", ".join(str(item.get("label") or item.get("make_id") or "segment") for item in segments), + len(segments) if full_pass_mode else min(MOBILEDE_RUNTIME_INITIAL_TASKS, len(segments)), + 0 if full_pass_mode else max(0, len(segments) - min(MOBILEDE_RUNTIME_INITIAL_TASKS, len(segments))), + "full-pass" if full_pass_mode else "incremental", + ", ".join(_mobilede_short_segment_label(item) for item in segments[: min(5, len(segments))]), ) return { "status": "queued", @@ -2271,148 +3068,213 @@ def mobilede_sync_search_task( task_id = self.request.id or "unknown" redis_client = _get_redis() settings = Settings() + segment_runtime_key = _mobilede_task_segment_key( + segment=segment, + search_url=search_url, + make_id=make_id, + model_id=model_id, + price_min=price_min, + price_max=price_max, + year_min=year_min, + year_max=year_max, + mileage_min=mileage_min, + mileage_max=mileage_max, + ) + segment_lock_key = _mobilede_segment_lock_key(segment_runtime_key) + lock_owner = f"{task_id}:{uuid.uuid4().hex}" + lock_ttl = max(300, _sync_listing_lock_ttl_seconds()) + lock_acquired = _acquire_lock(redis_client, segment_lock_key, lock_owner, lock_ttl) + if not lock_acquired: + logger.info("mobile.de sync skipped: segment already running key=%s", segment_runtime_key) + return {"status": "skipped", "reason": "segment_already_running", "segment_key": segment_runtime_key} + heartbeat_stop: Event | None = None + heartbeat_thread: Thread | None = None + watchdog_stop: Event | None = None + watchdog_thread: Thread | None = None + stall_timeout = max(120, int(settings.celery.task_stall_timeout_seconds)) + progress_ttl = max(lock_ttl + 120, stall_timeout + 120) runtime_config = RuntimeConfig.from_file(settings.runtime_config_file) - if only_new is None and runtime_config.sync.only_new is not None: - only_new = runtime_config.sync.only_new - runtime_segments_enabled = False - if segment is None and not make_id and not model_id: - reserved_segment = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new) - if reserved_segment is not None: - segment_index, segment = reserved_segment + try: + heartbeat_stop, heartbeat_thread = _start_lock_heartbeat( + redis_client, + segment_lock_key, + lock_owner, + lock_ttl, + ) + watchdog_stop, watchdog_thread = _start_stall_watchdog( + redis_client, + task_id=task_id, + stall_timeout_seconds=stall_timeout, + lock_key=segment_lock_key, + lock_owner=lock_owner, + ) + _clear_mobilede_followup_pending(redis_client, segment_key=segment_runtime_key) + if only_new is None and runtime_config.sync.only_new is not None: + only_new = runtime_config.sync.only_new + bootstrap_done = _mobilede_bootstrap_done(redis_client) + guarded_only_new = _mobilede_force_full_scan_only_new(only_new, redis_client=redis_client) + if only_new is True and guarded_only_new is False: + logger.info("mobile.de full-pass mode: forcing only_new=False in sync task") + only_new = guarded_only_new + runtime_segments_enabled = False + if segment is None and not make_id and not model_id: + reserved_segment = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new) + if reserved_segment is not None: + segment_index, segment = reserved_segment + runtime_segments_enabled = True + else: + if not redis_client.get(MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY): + mobilede_sync_runtime_segments_task.apply_async( + kwargs={ + "lane": lane, + "delay_seconds": delay_seconds, + "use_cursor": use_cursor, + "continuous": bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED), + }, + queue=MOBILEDE_SYNC_QUEUE, + ) + logger.info("mobile.de runtime segments are not ready; deferring sync task") + raise self.retry(countdown=30) + elif segment is not None: runtime_segments_enabled = True - else: - if not redis_client.get(MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY): - mobilede_sync_runtime_segments_task.apply_async( - kwargs={ - "lane": lane, - "delay_seconds": delay_seconds, - "use_cursor": use_cursor, - "continuous": bool(continuous if continuous is not None else MOBILEDE_CONTINUOUS_SYNC_ENABLED), - }, - queue=MOBILEDE_SYNC_QUEUE, - ) - logger.info("mobile.de runtime segments are not ready; deferring sync task") - raise self.retry(countdown=30) - elif segment is not None: - runtime_segments_enabled = True - if segment: - search_url = search_url or str(segment.get("search_url") or segment.get("listing_url") or "").strip() or None - make_id = make_id or str(segment.get("make_id") or "").strip() or None - model_id = model_id or str(segment.get("model_id") or "").strip() or None - price_min = price_min or str(segment.get("price_min") or "").strip() or None - price_max = price_max or str(segment.get("price_max") or "").strip() or None - year_min = year_min or str(segment.get("year_min") or "").strip() or None - year_max = year_max or str(segment.get("year_max") or "").strip() or None - mileage_min = mileage_min or str(segment.get("mileage_min") or "").strip() or None - mileage_max = mileage_max or str(segment.get("mileage_max") or "").strip() or None - if segment.get("only_new") is not None: - only_new = bool(segment.get("only_new")) - start_page = int(segment.get("start_page") or start_page or 1) - if segment.get("max_pages") is not None: - max_pages = int(segment.get("max_pages") or max_pages) - if _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: + if segment: + search_url = search_url or str(segment.get("search_url") or segment.get("listing_url") or "").strip() or None + make_id = make_id or str(segment.get("make_id") or "").strip() or None + model_id = model_id or str(segment.get("model_id") or "").strip() or None + price_min = price_min or str(segment.get("price_min") or "").strip() or None + price_max = price_max or str(segment.get("price_max") or "").strip() or None + year_min = year_min or str(segment.get("year_min") or "").strip() or None + year_max = year_max or str(segment.get("year_max") or "").strip() or None + mileage_min = mileage_min or str(segment.get("mileage_min") or "").strip() or None + mileage_max = mileage_max or str(segment.get("mileage_max") or "").strip() or None + if segment.get("only_new") is not None: + only_new = bool(segment.get("only_new")) + start_page = int(segment.get("start_page") or start_page or 1) + if segment.get("max_pages") is not None: + max_pages = int(segment.get("max_pages") or max_pages) + if only_new and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: + start_page = 1 + max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) + elif make_id or model_id: + resolved_segment = _find_mobilede_runtime_segment( + settings, + search_url=search_url, + make_id=make_id, + model_id=model_id, + ) + if resolved_segment is not None: + segment = resolved_segment + runtime_segments_enabled = True + elif search_url: + resolved_segment = _find_mobilede_runtime_segment( + settings, + search_url=search_url, + make_id=make_id, + model_id=model_id, + ) + if resolved_segment is not None: + segment = resolved_segment + runtime_segments_enabled = True + + if only_new and segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: start_page = 1 max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) - elif make_id or model_id: - resolved_segment = _find_mobilede_runtime_segment( - settings, - search_url=search_url, - make_id=make_id, - model_id=model_id, - ) - if resolved_segment is not None: - segment = resolved_segment - runtime_segments_enabled = True - elif search_url: - resolved_segment = _find_mobilede_runtime_segment( - settings, - search_url=search_url, - make_id=make_id, - model_id=model_id, - ) - if resolved_segment is not None: - segment = resolved_segment - runtime_segments_enabled = True + use_cursor = False - if segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: - start_page = 1 - max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) - use_cursor = False + only_new = _mobilede_force_full_scan_only_new(only_new, redis_client=redis_client) - strict_first_pass_mode = _mobilede_is_strict_first_pass_mode(redis_client, segment=segment, only_new=only_new) - cycle_id: str | None = None - if strict_first_pass_mode: - segment, segment_index, cycle_id = _mobilede_reserve_strict_first_pass_segment( - redis_client, - settings, - segment=segment, - segment_index=segment_index, - ) - start_page = 1 - max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) - use_cursor = False + strict_first_pass_mode = _mobilede_is_strict_first_pass_mode(redis_client, segment=segment, only_new=only_new) + cycle_id: str | None = None + if strict_first_pass_mode: + segment, segment_index, cycle_id = _mobilede_reserve_strict_first_pass_segment( + redis_client, + settings, + segment=segment, + segment_index=segment_index, + ) + start_page = 1 + max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) + use_cursor = False - sort_by: str | None = None - sort_order: str | None = None - if only_new and MOBILEDE_ONLY_NEW_NEWEST_FIRST: - sort_by = "doc" - sort_order = "down" - search_url = _mobilede_apply_newest_sort_to_url(search_url) + sort_by: str | None = None + sort_order: str | None = None + if only_new and MOBILEDE_ONLY_NEW_NEWEST_FIRST: + sort_by = "doc" + sort_order = "down" + search_url = _mobilede_apply_newest_sort_to_url(search_url) - cursor_key = _mobilede_cursor_key(segment) - if strict_first_pass_mode and cycle_id: - cursor_key = _mobilede_cycle_cursor_key(cursor_key, cycle_id) - actual_start_page, actual_end_page = _reserve_mobilede_page_window( - redis_client, - requested_start_page=start_page, - page_window_size=max_pages, - use_cursor=use_cursor, - cursor_key=cursor_key, - ) - if use_cursor and MOBILEDE_SKIP_EMPTY_WINDOW and actual_start_page > 100: - _reset_mobilede_page_cursor(redis_client, cursor_key=cursor_key, next_start_page=1) + cursor_key = _mobilede_cursor_key(segment) + if strict_first_pass_mode and cycle_id: + cursor_key = _mobilede_cycle_cursor_key(cursor_key, cycle_id) actual_start_page, actual_end_page = _reserve_mobilede_page_window( redis_client, - requested_start_page=1, + requested_start_page=start_page, page_window_size=max_pages, use_cursor=use_cursor, cursor_key=cursor_key, ) - logger.info( - "mobile.de cursor wrapped before empty window: runtime=%s pages=%s-%s", - _mobilede_segment_label(segment), - actual_start_page, - actual_end_page, + if use_cursor and MOBILEDE_SKIP_EMPTY_WINDOW and actual_start_page > 100: + _reset_mobilede_page_cursor(redis_client, cursor_key=cursor_key, next_start_page=1) + actual_start_page, actual_end_page = _reserve_mobilede_page_window( + redis_client, + requested_start_page=1, + page_window_size=max_pages, + use_cursor=use_cursor, + cursor_key=cursor_key, + ) + logger.info( + "mobile.de cursor wrapped before empty window: runtime=%s pages=%s-%s", + _mobilede_segment_label(segment), + actual_start_page, + actual_end_page, + ) + _update_task_progress( + redis_client, + task_id=task_id, + stage="mobilede_sync_started", + ttl_seconds=progress_ttl, + start_page=actual_start_page, + end_page=actual_end_page, + requested_start_page=start_page, + max_pages=max_pages, + use_cursor=use_cursor, + segment_index=segment_index, + segment_label=(segment or {}).get("label") if segment else None, ) - _update_task_progress( - redis_client, - task_id=task_id, - stage="mobilede_sync_started", - ttl_seconds=3600, - start_page=actual_start_page, - end_page=actual_end_page, - requested_start_page=start_page, - max_pages=max_pages, - use_cursor=use_cursor, - segment_index=segment_index, - segment_label=(segment or {}).get("label") if segment else None, - ) - try: if continuous is None: continuous = MOBILEDE_CONTINUOUS_SYNC_ENABLED segment_label = _mobilede_segment_label(segment) make_name = _mobilede_segment_make(segment, make_id) model_name = _mobilede_segment_model(segment, model_id) + incremental_progress_mode = bool(strict_first_pass_mode and cycle_id) + if incremental_progress_mode: + progress_cycle_id, progress_done, progress_total, progress_left = _mobilede_incremental_cycle_progress( + redis_client, + cycle_id=cycle_id, + ) + progress_scope = f"incremental cycle {progress_cycle_id}" + else: + progress_done, progress_total, progress_left = _mobilede_bootstrap_progress(redis_client) + progress_scope = "bootstrap" + progress_percent = _mobilede_bootstrap_percent(progress_done, progress_total) + total_listings, total_unique, total_inserted, total_updated, _total_images = _mobilede_bootstrap_cars_totals(redis_client) logger.info( - "mobile.de sync started: runtime=%s filter=%s pages=%s-%s max_pages=%s use_cursor=%s continuous=%s sort=%s:%s", - segment_label, - _mobilede_filter_source(segment, search_url), + "mobile.de segment start: segment=%s pages=%s-%s max_pages=%s mode=%s progress=%s/%s left=%s (%.1f%%) total_cars: listings=%s unique=%s inserted=%s updated=%s filter=%s sort=%s:%s", + _mobilede_short_segment_label(segment), actual_start_page, actual_end_page, max_pages, - use_cursor, - continuous, + progress_scope, + progress_done, + progress_total, + progress_left, + progress_percent, + total_listings, + total_unique, + total_inserted, + total_updated, + _mobilede_filter_source(segment, search_url), sort_by, sort_order, ) @@ -2498,10 +3360,49 @@ def mobilede_sync_search_task( listings=int(result.get("listing_count", 0) or 0), ) listing_count = int(result.get("listing_count", 0) or 0) - if use_cursor and _mobilede_segment_scan_complete(redis_client, segment): + unique_count = int(result.get("unique_listing_count", 0) or 0) + updated_count = int(result.get("upsert", {}).get("updated", 0) or 0) + images_count = int(result.get("upsert", {}).get("images_upserted", 0) or 0) + bootstrap_progress: tuple[int, int, int, int, int, int, int] | None = None + overflow_segments_added = 0 + segment_max_pages = int((segment or {}).get("max_pages") or max_pages or MOBILEDE_MAX_PAGE_NUMBER) + segment_scan_complete = bool( + (use_cursor and _mobilede_segment_scan_complete(redis_client, segment)) + or ( + not use_cursor + and segment is not None + and actual_start_page <= 1 + and actual_end_page >= segment_max_pages + ) + ) + if segment_scan_complete: + overflow_segments_added = _mobilede_try_expand_overflow_segment( + redis_client, + settings, + segment=segment, + listing_count=listing_count, + unique_count=unique_count, + max_pages=max_pages, + segment_end_page=actual_end_page, + ) + if segment_scan_complete: total_segments_raw = redis_client.get(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY) total_segments = int(total_segments_raw) if total_segments_raw else None - _mark_mobilede_bootstrap_segment_done(redis_client, segment, total_segments) + bootstrap_progress = _mark_mobilede_bootstrap_segment_done( + redis_client, + segment, + total_segments, + listings=listing_count, + unique=unique_count, + inserted=inserted_count, + updated=updated_count, + images=images_count, + ) + if bootstrap_progress is not None: + try: + redis_client.srem(MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY, _mobilede_segment_fingerprint(segment)) + except Exception: + logger.debug("Failed to release bootstrap dispatched marker for segment", exc_info=True) if use_cursor and listing_count == 0: _reset_mobilede_page_cursor(redis_client, cursor_key=cursor_key, next_start_page=1) logger.debug( @@ -2534,21 +3435,83 @@ def mobilede_sync_search_task( result.get("upsert", {}).get("inserted", 0), result.get("upsert", {}).get("updated", 0), ) + if incremental_progress_mode: + progress_cycle_id, progress_done, progress_total, progress_left = _mobilede_incremental_cycle_progress( + redis_client, + cycle_id=cycle_id, + ) + progress_scope = f"incremental cycle {progress_cycle_id}" + else: + progress_done, progress_total, progress_left = ( + bootstrap_progress[:3] if bootstrap_progress else _mobilede_bootstrap_progress(redis_client) + ) + progress_scope = "bootstrap" logger.info( - "mobile.de sync completed: runtime=%s filter=%s pages=%s-%s listings=%s unique=%s inserted=%s updated=%s images=%s run_id=%s", - segment_label, - _mobilede_filter_source(segment, search_url), + "mobile.de segment result: segment=%s pages=%s-%s cars: listings=%s unique=%s inserted=%s updated=%s images=%s mode=%s progress=%s/%s left=%s run_id=%s overflow_added=%s", + _mobilede_short_segment_label(segment), actual_start_page, actual_end_page, - int(result.get("listing_count", 0) or 0), - int(result.get("unique_listing_count", 0) or 0), + listing_count, + unique_count, inserted_count, - int(result.get("upsert", {}).get("updated", 0) or 0), - int(result.get("upsert", {}).get("images_upserted", 0) or 0), + updated_count, + images_count, + progress_scope, + progress_done, + progress_total, + progress_left, result.get("run_id"), + overflow_segments_added, ) if continuous: - incremental_mode = bool(segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP) + progress_done_now, progress_total_now, progress_left_now = _mobilede_bootstrap_progress(redis_client) + incremental_mode = bool(only_new and segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP) + if ( + MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED + and progress_total_now > 0 + and progress_done_now >= progress_total_now + and not incremental_mode + ): + should_start_incremental = bool( + runtime_config.sync.only_new is True + and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP + ) + if should_start_incremental: + if _try_queue_mobilede_incremental_transition( + redis_client, + lane=lane, + delay_seconds=delay_seconds, + use_cursor=False, + ): + logger.info( + "mobile.de bootstrap->incremental transition queued: bootstrap=%s/%s runtime=%s", + progress_done_now, + progress_total_now, + segment_label, + ) + else: + logger.info( + "mobile.de bootstrap->incremental transition already pending: bootstrap=%s/%s runtime=%s", + progress_done_now, + progress_total_now, + segment_label, + ) + else: + logger.info( + "mobile.de follow-up stopped: bootstrap completed (%s/%s), runtime=%s", + progress_done_now, + progress_total_now, + segment_label, + ) + return _mobilede_task_result_summary( + result=result, + segment=segment, + start_page=actual_start_page, + end_page=actual_end_page, + make_name=make_name, + model_name=model_name, + ) + bootstrap_rotation_mode = bool(segment and runtime_rotation and MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED and not _mobilede_bootstrap_done(redis_client)) next_start_page = 1 if incremental_mode else (actual_end_page + 1 if listing_count > 0 else 1) next_max_pages = min(max_pages, MOBILEDE_INCREMENTAL_PAGE_WINDOW) if incremental_mode else max_pages followup_kwargs = { @@ -2591,7 +3554,7 @@ def mobilede_sync_search_task( "segment_index": next_segment_index, } ) - if _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: + if only_new and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP: followup_kwargs["start_page"] = 1 followup_kwargs["max_pages"] = min(int(followup_kwargs["max_pages"]), MOBILEDE_INCREMENTAL_PAGE_WINDOW) followup_kwargs["use_cursor"] = False @@ -2609,26 +3572,78 @@ def mobilede_sync_search_task( elif segment is not None and runtime_segments_enabled: followup_kwargs["segment"] = None followup_kwargs["segment_index"] = None - mobilede_sync_search_task.apply_async( - kwargs=followup_kwargs, - queue=MOBILEDE_SYNC_QUEUE, - countdown=MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, - ) - logger.debug( - "mobilede_sync_search_task queued follow-up: segment=%s next_start_page=%s delay=%ss use_cursor=%s", - (followup_kwargs.get("segment") or {}).get("label") if isinstance(followup_kwargs.get("segment"), dict) else None, - next_start_page, - MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, - use_cursor, - ) - logger.info( - "mobile.de sync next window queued: runtime=%s filter=%s next_pages=%s-%s delay=%ss", - segment_label, - _mobilede_filter_source(segment, search_url), - next_start_page, - next_start_page + int(followup_kwargs["max_pages"]) - 1, - MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, - ) + if bootstrap_rotation_mode and "segment" not in followup_kwargs: + logger.info("mobile.de bootstrap follow-up skipped: no fresh runtime segment available after %s", segment_label) + mobilede_sync_runtime_segments_task.apply_async( + kwargs={ + "lane": lane, + "delay_seconds": delay_seconds, + "use_cursor": use_cursor, + "continuous": True, + }, + queue=MOBILEDE_SYNC_QUEUE, + countdown=5, + ) + logger.info( + "mobile.de bootstrap recovery queued: runtime=%s delay=%ss", + segment_label, + 5, + ) + return _mobilede_task_result_summary( + result=result, + segment=segment, + start_page=actual_start_page, + end_page=actual_end_page, + make_name=make_name, + model_name=model_name, + ) + followup_segment = followup_kwargs.get("segment") + followup_segment_key = _mobilede_segment_fingerprint(followup_segment) if isinstance(followup_segment, dict) else segment_runtime_key + if _try_set_mobilede_followup_pending( + redis_client, + segment_key=followup_segment_key, + ttl_seconds=max(lock_ttl, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS + 300), + ): + mobilede_sync_search_task.apply_async( + kwargs=followup_kwargs, + queue=MOBILEDE_SYNC_QUEUE, + countdown=MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, + ) + logger.debug( + "mobilede_sync_search_task queued follow-up: segment=%s next_start_page=%s delay=%ss use_cursor=%s", + (followup_kwargs.get("segment") or {}).get("label") if isinstance(followup_kwargs.get("segment"), dict) else None, + next_start_page, + MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, + use_cursor, + ) + logger.info( + "mobile.de sync next window queued: runtime=%s filter=%s next_pages=%s-%s delay=%ss", + segment_label, + _mobilede_filter_source(segment, search_url), + next_start_page, + next_start_page + int(followup_kwargs["max_pages"]) - 1, + MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, + ) + else: + if _try_reset_stale_mobilede_followup_pending(redis_client, segment_key=followup_segment_key) and _try_set_mobilede_followup_pending( + redis_client, + segment_key=followup_segment_key, + ttl_seconds=max(lock_ttl, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS + 300), + ): + mobilede_sync_search_task.apply_async( + kwargs=followup_kwargs, + queue=MOBILEDE_SYNC_QUEUE, + countdown=MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, + ) + logger.warning( + "mobile.de stale follow-up recovered: runtime=%s next_pages=%s-%s delay=%ss", + segment_label, + next_start_page, + next_start_page + int(followup_kwargs["max_pages"]) - 1, + MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS, + ) + else: + logger.info("mobile.de follow-up already pending for segment=%s", followup_segment_key) return _mobilede_task_result_summary( result=result, segment=segment, @@ -2688,19 +3703,27 @@ def mobilede_sync_search_task( if segment is not None: followup_kwargs["segment"] = segment followup_kwargs["segment_index"] = segment_index - mobilede_sync_search_task.apply_async( - kwargs=followup_kwargs, - queue=MOBILEDE_SYNC_QUEUE, - countdown=max(300, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS * 4), - ) - logger.warning( - "mobile.de delayed retry queued after network issue: runtime=%s filter=%s pages=%s-%s delay=%ss", - _mobilede_segment_label(segment), - _mobilede_filter_source(segment, search_url), - actual_start_page, - actual_end_page, - max(300, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS * 4), - ) + delayed_retry = max(300, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS * 4) + if _try_set_mobilede_followup_pending( + redis_client, + segment_key=segment_runtime_key, + ttl_seconds=max(lock_ttl, delayed_retry + 300), + ): + mobilede_sync_search_task.apply_async( + kwargs=followup_kwargs, + queue=MOBILEDE_SYNC_QUEUE, + countdown=delayed_retry, + ) + logger.warning( + "mobile.de delayed retry queued after network issue: runtime=%s filter=%s pages=%s-%s delay=%ss", + _mobilede_segment_label(segment), + _mobilede_filter_source(segment, search_url), + actual_start_page, + actual_end_page, + delayed_retry, + ) + else: + logger.info("mobile.de delayed retry already pending for segment=%s", segment_runtime_key) return { "status": "network_error_deferred", "runtime": _mobilede_segment_label(segment), @@ -2723,6 +3746,17 @@ def mobilede_sync_search_task( exc_info=True, ) raise self.retry(exc=exc) + finally: + if watchdog_stop is not None: + watchdog_stop.set() + if watchdog_thread is not None: + watchdog_thread.join(timeout=5) + if heartbeat_stop is not None: + heartbeat_stop.set() + if heartbeat_thread is not None: + heartbeat_thread.join(timeout=max(1.0, min(5.0, lock_ttl / 10))) + _clear_task_progress(redis_client, task_id) + _release_lock_if_owner(redis_client, segment_lock_key, lock_owner) @shared_task(