from __future__ import annotations # Планирование сегментов вынесено сюда, чтобы `tasks.py` оставался короче. # Модуль использует хелперы из `tasks.py`. from .tasks import * # noqa: F401,F403 from .tasks import ( _mobilede_make_segment_url, _mobilede_segment_fingerprint, _mobilede_segment_key, _mobilede_segment_label, _mobilede_short_segment_label, _mobilede_short_segment_ref, _mobilede_should_skip_dynamic_segment, _mobilede_should_skip_planned_segment, _mobilede_site_make_options_cache, _mobilede_try_mark_bootstrap_segment_dispatched, _mobilede_url_query_values, ) _MOBILEDE_RUNTIME_PLAN_VERSION = 2 def _mobilede_price_ranges() -> list[tuple[int, int | None]]: raw_ranges = os.getenv("MOBILEDE_PRICE_RANGES", "").strip() if raw_ranges: parsed: list[tuple[int, int | None]] = [] for raw_item in raw_ranges.split(","): item = raw_item.strip() if not item: continue left, _, right = item.partition(":") try: min_value = int(left.strip()) if left.strip() else 1 max_value = int(right.strip()) if right.strip() else None parsed.append((min_value, max_value)) except ValueError: logger.warning("Invalid MOBILEDE_PRICE_RANGES item ignored: %s", item) if parsed: return parsed if MOBILEDE_COMPACT_SEGMENTS: return [ (1, 5000), (5001, 10000), (10001, 15000), (15001, 20000), (20001, 30000), (30001, 50000), (50001, 75000), (75001, 100000), (100001, 150000), (150001, None), ] return [(1, 500), (500, 1000), (1001, 1500), (1501, 2000), (2001, 2500), (2501, 3000), (3001, 4000), (4001, 5000), (5001, 7500), (7501, 10000), (10001, 12500), (12501, 15000), (15001, 17500), (17501, 20000), (20001, 25000), (25001, 30000), (30001, 40000), (40001, 50000), (50001, 75000), (75001, 100000), (100001, 150000), (150001, None)] def _mobilede_price_ranges_for_segment(segment: dict[str, object] | None = None) -> list[tuple[int, int | None]]: del segment if MOBILEDE_COMPACT_SEGMENTS and os.getenv("MOBILEDE_DENSE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}: # Плотные корзины включаются для всех марок из пользовательской ссылки. # Так покрытие не зависит от конкретных make_id и не теряет хвосты за лимитом 50 страниц. return [ (1, 5000), (5001, 10000), (10001, 15000), (15001, 20000), (20001, 25000), (25001, 30000), (30001, 35000), (35001, 40000), (40001, 50000), (50001, 60000), (60001, 80000), (80001, 100000), (100001, 150000), (150001, None), ] return _mobilede_price_ranges() def _mobilede_filtered_url_uses_adaptive_plan() -> bool: return os.getenv("MOBILEDE_FILTERED_URL_ADAPTIVE_PLAN", "false").strip().lower() in {"1", "true", "yes", "on"} def _mobilede_skip_late_overflow_children_during_bootstrap() -> bool: return os.getenv("MOBILEDE_SKIP_LATE_OVERFLOW_CHILDREN_DURING_BOOTSTRAP", "true").strip().lower() in {"1", "true", "yes", "on"} def _mobilede_year_ranges() -> list[tuple[int | None, int | None]]: if MOBILEDE_COMPACT_SEGMENTS: return [(None, 2009), (2010, 2017), (2018, 2022), (2023, 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_year_ranges_for_segment_price( segment: dict[str, object] | None, price_min: int, price_max: int | None, ) -> list[tuple[int | None, int | None]]: del segment if not (MOBILEDE_COMPACT_SEGMENTS and os.getenv("MOBILEDE_DENSE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}): return _mobilede_year_ranges_for_price(price_min, price_max) 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 <= 30000: return [(None, 2009), (2010, 2013), (2014, 2017), (2018, 2020), (2021, 2022), (2023, 2024), (2025, None)] if upper_bound <= 50000: return [(None, 2009), (2010, 2017), (2018, 2020), (2021, 2022), (2023, 2024), (2025, None)] return _mobilede_year_ranges_for_price(price_min, price_max) def _mobilede_interleave_segments_by_make(segments: list[dict[str, object]]) -> list[dict[str, object]]: """Mix make blocks so an early bootstrap pass covers the whole source URL.""" groups: dict[str, list[dict[str, object]]] = {} order: list[str] = [] passthrough: list[dict[str, object]] = [] for segment in segments: make_id = str(segment.get("make_id") or "").strip() if not make_id: passthrough.append(segment) continue if make_id not in groups: groups[make_id] = [] order.append(make_id) groups[make_id].append(segment) if len(order) <= 1: return segments mixed: list[dict[str, object]] = [] max_len = max(len(items) for items in groups.values()) for index in range(max_len): for make_id in order: items = groups[make_id] if index < len(items): mixed.append(items[index]) mixed.extend(passthrough) return mixed def _mobilede_mileage_ranges() -> list[tuple[int | None, int | None]]: if MOBILEDE_COMPACT_SEGMENTS: return [(None, 100000), (100001, 200000), (200001, 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 if MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE: return True # Делим по пробегу только плотные корзины. 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 return ( 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_target_results_band() -> tuple[int, int, int]: target = int(min(MOBILEDE_SEGMENT_TARGET_RESULTS, _mobilede_overflow_threshold(MOBILEDE_MAX_PAGE_NUMBER))) lower = max(MOBILEDE_RESULTS_PER_PAGE, int(target * MOBILEDE_SEGMENT_TARGET_MIN_RATIO)) upper = max(lower, min(target, int(target * MOBILEDE_SEGMENT_TARGET_MAX_RATIO))) tiny = max(MOBILEDE_RESULTS_PER_PAGE, int(target * MOBILEDE_SEGMENT_TINY_RATIO)) return target, lower, tiny def _mobilede_range_value(min_value: int | None, max_value: int | None) -> str: return f"{min_value or ''}:{max_value or ''}" def _mobilede_price_label(price_min: int, price_max: int | None) -> str: return f"price={price_min}-{price_max}" if price_max is not None else f"price={price_min}+" def _mobilede_year_label(year_min: int | None, year_max: int | None) -> str: if year_min is None: return f"year<={year_max}" if year_max is None: return f"year>={year_min}" return f"year={year_min}-{year_max}" def _mobilede_mileage_label(mileage_min: int | None, mileage_max: int | None) -> str: if mileage_min is None: return f"km<={mileage_max}" if mileage_max is None: return f"km>={mileage_min}" 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_segment_total_results(segment: dict[str, object] | None) -> int | None: if not segment: return None return _mobilede_parse_optional_int(segment.get("total_results")) def _mobilede_prune_overflow_parent_segments(segments: list[dict[str, object]]) -> list[dict[str, object]]: """Keep only leaf segments once overflow children exist for a parent.""" parent_keys = { str(item.get("overflow_parent") or "").strip() for item in segments if isinstance(item, dict) and str(item.get("overflow_parent") or "").strip() } if not parent_keys: return [dict(item) for item in segments if isinstance(item, dict)] pruned: list[dict[str, object]] = [] removed = 0 for item in segments: if not isinstance(item, dict): continue if _mobilede_segment_fingerprint(item) in parent_keys: removed += 1 continue pruned.append(dict(item)) if removed: logger.info( "mobile.de overflow parent segments pruned: before=%s after=%s removed=%s", len(segments), len(pruned), removed, ) return pruned def _mobilede_learned_segments_file() -> str: return os.getenv("MOBILEDE_LEARNED_SEGMENTS_FILE", "/data/mobilede_runtime_segments.json").strip() def _mobilede_segments_source_fingerprint(segments: list[dict[str, object]]) -> str: source_urls = [ str(segment.get("source_search_url") or segment.get("search_url") or segment.get("listing_url") or "").strip() for segment in segments if isinstance(segment, dict) ] source = json.dumps(sorted(url for url in source_urls if url), ensure_ascii=False, separators=(",", ":")) return hashlib.sha256(source.encode("utf-8")).hexdigest()[:16] def _mobilede_load_learned_runtime_segments(source_segments: list[dict[str, object]]) -> list[dict[str, object]] | None: path = _mobilede_learned_segments_file() if not path: return None try: with open(path, "r", encoding="utf-8") as fh: payload = json.load(fh) if not isinstance(payload, dict): return None if payload.get("plan_version") != _MOBILEDE_RUNTIME_PLAN_VERSION: return None expected = _mobilede_segments_source_fingerprint(source_segments) if str(payload.get("source_fingerprint") or "") != expected: logger.info( "mobile.de learned segments ignored: source fingerprint changed expected=%s actual=%s", expected, payload.get("source_fingerprint"), ) return None raw_segments = payload.get("segments") if not isinstance(raw_segments, list): return None learned = _mobilede_prune_overflow_parent_segments([dict(item) for item in raw_segments if isinstance(item, dict)]) if not learned: return None logger.info("mobile.de learned runtime segments loaded: file=%s count=%s", path, len(learned)) return learned except FileNotFoundError: return None except Exception: logger.warning("mobile.de learned runtime segments load failed: file=%s", path, exc_info=True) return None def _mobilede_save_learned_runtime_segments( source_segments: list[dict[str, object]], runtime_segments: list[dict[str, object]], ) -> None: path = _mobilede_learned_segments_file() if not path or not runtime_segments: return try: os.makedirs(os.path.dirname(path) or ".", exist_ok=True) pruned_segments = _mobilede_prune_overflow_parent_segments(runtime_segments) payload = { "plan_version": _MOBILEDE_RUNTIME_PLAN_VERSION, "source_fingerprint": _mobilede_segments_source_fingerprint(source_segments), "updated_at": datetime.now(timezone.utc).isoformat(), "segments": pruned_segments, } tmp_path = f"{path}.tmp" with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(payload, fh, ensure_ascii=False, separators=(",", ":")) os.replace(tmp_path, path) logger.info("mobile.de learned runtime segments saved: file=%s count=%s", path, len(pruned_segments)) except Exception: logger.warning("mobile.de learned runtime segments save failed: file=%s", path, exc_info=True) def _mobilede_normalize_make_name(value: str) -> str: text = unicodedata.normalize("NFKD", str(value or "")) ascii_text = text.encode("ascii", "ignore").decode("ascii") return re.sub(r"[^a-z0-9]+", "", ascii_text.casefold()) def _mobilede_runtime_filter_brands(settings: Settings) -> tuple[str, ...]: runtime_config = RuntimeConfig.from_file(settings.runtime_config_file) return tuple(runtime_config.filters.include.brands) def _mobilede_make_alias_candidates(value: str) -> tuple[str, ...]: normalized = _mobilede_normalize_make_name(value) aliases = _MOBILEDE_RUNTIME_BRAND_ALIASES.get(normalized, ()) candidates = [normalized, *aliases] deduped: list[str] = [] for candidate in candidates: if candidate and candidate not in deduped: deduped.append(candidate) return tuple(deduped) def _mobilede_strip_query_keys(search_url: str, keys: set[str]) -> str: parts = urlsplit(search_url) query_items = [ (key, value) for key, value in parse_qsl(parts.query, keep_blank_values=True) if key not in keys ] return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query_items, doseq=True), parts.fragment)) def _mobilede_refdata_make_keys() -> dict[str, str]: global _mobilede_refdata_make_keys_cache now = time.monotonic() if _mobilede_refdata_make_keys_cache is not None: cached_at, cached = _mobilede_refdata_make_keys_cache if now - cached_at < MOBILEDE_SITE_MAKE_OPTIONS_CACHE_TTL_SECONDS and cached: return cached try: response = requests.get( MOBILEDE_REFDATA_CAR_MAKES_URL, headers={"Accept": "application/xml", "Accept-Language": "en"}, timeout=60, ) response.raise_for_status() root = ElementTree.fromstring(response.text) make_keys: dict[str, str] = {} for item in root.iter(): key = str(item.attrib.get("key") or "").strip() normalized = _mobilede_normalize_make_name(key) if key and normalized and normalized not in make_keys: make_keys[normalized] = key for alias in _MOBILEDE_REFDATA_MAKE_KEY_ALIASES.get(normalized, ()): if alias and alias not in make_keys: make_keys[alias] = key _mobilede_refdata_make_keys_cache = (now, make_keys) return make_keys except Exception: logger.warning("mobile.de refdata make keys load failed", exc_info=True) if _mobilede_refdata_make_keys_cache is not None: return _mobilede_refdata_make_keys_cache[1] return {} def _mobilede_parse_site_make_options(html: str) -> dict[str, str]: official_make_keys = _mobilede_refdata_make_keys() if not official_make_keys: return {} options: dict[str, str] = {} for raw_label, raw_value in _MOBILEDE_SITE_MAKE_OPTION_RE.findall(html): label = str(raw_label or "").strip() value = str(raw_value or "").strip() normalized = _mobilede_normalize_make_name(label) if not label or not value.isdigit() or not normalized: continue if normalized not in official_make_keys: continue options.setdefault(label, value) return options def _mobilede_fetch_site_make_options(search_url: str) -> dict[str, str]: base_url = _mobilede_strip_query_keys(search_url, {"ms"}) now = time.monotonic() cached = _mobilede_site_make_options_cache.get(base_url) if cached is not None: cached_at, options = cached if now - cached_at < MOBILEDE_SITE_MAKE_OPTIONS_CACHE_TTL_SECONDS and options: return options client = MobileDeClient.for_worker(delay_seconds=0) html = client.fetch_html(base_url, timeout=60) options = _mobilede_parse_site_make_options(html) if options: _mobilede_site_make_options_cache[base_url] = (now, options) return options def _mobilede_match_site_make_option( brand_name: str, make_options: dict[str, str], ) -> tuple[str, str] | None: option_by_normalized = { _mobilede_normalize_make_name(label): (label, value) for label, value in make_options.items() } for candidate in _mobilede_make_alias_candidates(brand_name): match = option_by_normalized.get(candidate) if match is not None: return match return None def _mobilede_source_segments_from_runtime_brands( settings: Settings, env_search_urls: list[str], ) -> list[dict[str, object]]: if not MOBILEDE_RUNTIME_BRAND_SEGMENTS_ENABLED or not env_search_urls: return [] runtime_brands = list(_mobilede_runtime_filter_brands(settings)) derived_segments: list[dict[str, object]] = [] seen_fingerprints: set[str] = set() missing_brands: set[str] = set() for index, search_url in enumerate(env_search_urls, start=1): base_url = _mobilede_strip_query_keys(search_url, {"ms"}) try: make_options = _mobilede_fetch_site_make_options(base_url) except Exception as exc: logger.warning("mobile.de site make options fetch failed: url=%s error=%s", base_url, exc) continue if not make_options: continue if runtime_brands: selected: list[tuple[str, str, str]] = [] for brand_name in runtime_brands: match = _mobilede_match_site_make_option(brand_name, make_options) if match is None: missing_brands.add(brand_name) continue site_label, make_token = match selected.append((brand_name, site_label, make_token)) else: selected = [(site_label, site_label, make_token) for site_label, make_token in sorted(make_options.items())] base_label = "Cars" if len(env_search_urls) == 1 else f"Cars {index}" for requested_brand, site_label, make_token in selected: item = { "label": f"{site_label} | ms={make_token}" if len(env_search_urls) == 1 else f"{base_label} | {site_label} | ms={make_token}", "make": site_label, "make_id": make_token, "search_url": _mobilede_make_segment_url(base_url, ms=make_token), "source_search_url": base_url, "start_page": 1, "max_pages": MOBILEDE_MAX_PAGE_NUMBER, "auto_segment": True, "runtime_brand": requested_brand, } fingerprint = _mobilede_segment_fingerprint(item) if fingerprint in seen_fingerprints: continue seen_fingerprints.add(fingerprint) derived_segments.append(item) if missing_brands: logger.warning( "mobile.de runtime brands skipped: missing=%s brands=%s", len(missing_brands), ", ".join(sorted(missing_brands)), ) if derived_segments: logger.info( "mobile.de runtime brand source segments ready: templates=%s requested_brands=%s segments=%s", len(env_search_urls), len(runtime_brands) if runtime_brands else len(derived_segments), len(derived_segments), ) return derived_segments def _mobilede_source_segments_from_settings(settings: Settings) -> list[dict[str, object]]: env_search_urls = settings.listing.filtered_search_urls if env_search_urls: runtime_brand_segments = _mobilede_source_segments_from_runtime_brands(settings, env_search_urls) if runtime_brand_segments: return runtime_brand_segments return [ { "label": "Cars" if len(env_search_urls) == 1 else f"Cars {index}", "search_url": search_url, "source_search_url": search_url, "start_page": 1, "max_pages": MOBILEDE_MAX_PAGE_NUMBER, } for index, search_url in enumerate(env_search_urls, start=1) ] runtime_config = RuntimeConfig.from_file(settings.runtime_config_file) return [segment.to_task_kwargs() for segment in runtime_config.mobilede.segments] 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 < MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN: 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: child_budget = max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)) step = max(MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN, 5000, min(50000, left)) ranges: list[tuple[int, int | None]] = [] current = left for _index in range(child_budget - 1): upper = current + step ranges.append((current, upper)) current = upper + 1 step = min(step * 2, 100000) ranges.append((current, None)) return ranges span = int(right) - int(left) if span < MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN: return [] child_count = min( max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)), max(2, span // MOBILEDE_OVERFLOW_MIN_PRICE_SPLIT_SPAN + 1), ) step = max(1, span // child_count) ranges = [] current = int(left) for index in range(child_count): upper = int(right) if index == child_count - 1 else min(int(right), current + step) if upper < current: break ranges.append((current, upper)) current = upper + 1 if current > int(right): break return ranges def _mobilede_split_mileage_ranges_for_overflow( mileage_min: int | None, mileage_max: int | None, ) -> list[tuple[int | None, int | None]]: if mileage_min is None and mileage_max is None: return [] if mileage_min is None and mileage_max is not None: right = int(mileage_max) if right <= 1: return [] if right <= 10000: return _mobilede_split_fine_mileage_ranges(None, right) pivot = right // 2 if pivot <= 0 or pivot >= right: return [] return [(None, pivot), (pivot + 1, right)] if mileage_min is not None and mileage_max is None: left = int(mileage_min) child_budget = max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)) step = max(25000, min(100000, left)) ranges: list[tuple[int | None, int | None]] = [] current = left for _index in range(child_budget - 1): upper = current + step ranges.append((current, upper)) current = upper + 1 ranges.append((current, None)) return ranges assert mileage_min is not None and mileage_max is not None left = int(mileage_min) right = int(mileage_max) span = right - left if span < 10000: return _mobilede_split_fine_mileage_ranges(left, right) child_count = min(max(2, int(MOBILEDE_OVERFLOW_MAX_CHILD_SEGMENTS)), max(2, span // 10000 + 1)) step = max(1, span // child_count) ranges = [] current = left for index in range(child_count): upper = right if index == child_count - 1 else min(right, current + step) if upper < current: break ranges.append((current, upper)) current = upper + 1 if current > right: break return ranges def _mobilede_root_mileage_ranges_for_overflow(max_children: int) -> list[tuple[int | None, int | None]]: child_budget = max(0, int(max_children)) if child_budget < 2: return [] if child_budget == 2: pivot = 150000 if MOBILEDE_COMPACT_SEGMENTS else 100000 return [(None, pivot), (pivot + 1, None)] if child_budget == 3: pivot = 150000 if MOBILEDE_COMPACT_SEGMENTS else 100000 low_cap = 75000 if MOBILEDE_COMPACT_SEGMENTS else 50000 return [(None, low_cap), (low_cap + 1, pivot), (pivot + 1, None)] if MOBILEDE_COMPACT_SEGMENTS: ranges: list[tuple[int | None, int | None]] = [ (None, 50000), (50001, 100000), (100001, 150000), (150001, 200000), (200001, None), ] return ranges[:child_budget] return _mobilede_mileage_ranges() def _mobilede_split_fine_mileage_ranges( mileage_min: int | None, mileage_max: int | None, ) -> list[tuple[int | None, int | None]]: if mileage_max is None: return [] left = int(mileage_min or 0) right = int(mileage_max) if right <= left: return [] span = right - left if span >= 5000: pivot = left + span // 2 elif span >= 1000: pivot = left + max(1, span // 2) else: return [] if pivot <= left or pivot >= right: return [] first_min = None if mileage_min is None and left == 0 else left return [(first_min, pivot), (pivot + 1, right)] def _mobilede_range_span(value_min: int | None, value_max: int | None) -> int | None: if value_min is None or value_max is None: return None return max(0, int(value_max) - int(value_min)) def _mobilede_should_avoid_tiny_mileage_split( *, depth: int, price_min: int | None, price_max: int | None, mileage_min: int | None, mileage_max: int | None, previous_split_kind: str, ) -> bool: price_span = _mobilede_range_span(price_min, price_max) mileage_span = _mobilede_range_span(mileage_min, mileage_max) if mileage_span is not None and mileage_span <= 25000: return True if previous_split_kind == "mileage" and mileage_span is not None and mileage_span <= 50000: return True if depth >= 2 and mileage_span is not None and mileage_span <= 100000: return True if price_span is not None and price_span <= 5000 and mileage_span is not None and mileage_span <= 100000: return True if price_span is not None and price_span <= 10000 and previous_split_kind == "mileage": return True return False 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_append_overflow_candidate_group( groups: list[tuple[str, list[dict[str, object]]]], kind: str, children: list[dict[str, object]], ) -> None: if not children: return fingerprints = tuple(_mobilede_segment_fingerprint(child) for child in children) if not all(fingerprints): return for existing_kind, existing_children in groups: existing_fingerprints = tuple(_mobilede_segment_fingerprint(child) for child in existing_children) if existing_kind == kind or existing_fingerprints == fingerprints: return groups.append((kind, children)) def _mobilede_build_overflow_candidate_groups( *, segment: dict[str, object], max_pages: int, ) -> list[tuple[str, 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)) candidate_groups: list[tuple[str, list[dict[str, object]]]] = [] has_mileage_filter = bool( str(segment.get("mileage_min") or "").strip() or str(segment.get("mileage_max") or "").strip() or "ml" in query_keys ) mileage_min = _mobilede_parse_optional_int(segment.get("mileage_min")) mileage_max = _mobilede_parse_optional_int(segment.get("mileage_max")) year_min = _mobilede_parse_optional_int(segment.get("year_min")) year_max = _mobilede_parse_optional_int(segment.get("year_max")) price_min = _mobilede_parse_optional_int(segment.get("price_min")) price_max = _mobilede_parse_optional_int(segment.get("price_max")) previous_split_kind = str(segment.get("overflow_split") or "").strip() mileage_splits = _mobilede_split_mileage_ranges_for_overflow(mileage_min, mileage_max) if mileage_splits: if _mobilede_should_avoid_tiny_mileage_split( depth=depth, price_min=price_min, price_max=price_max, mileage_min=mileage_min, mileage_max=mileage_max, previous_split_kind=previous_split_kind, ): mileage_splits = [] if not has_mileage_filter: child_segments: list[dict[str, object]] = [] for child_mileage_min, child_mileage_max in _mobilede_root_mileage_ranges_for_overflow(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(child_mileage_min, child_mileage_max), split_params={"ml": _mobilede_range_value(child_mileage_min, child_mileage_max)}, mileage_range=(child_mileage_min, child_mileage_max), ) ) _mobilede_append_overflow_candidate_group(candidate_groups, "root_mileage", child_segments) price_splits = _mobilede_split_price_ranges_for_overflow(price_min, price_max) if price_splits: child_segments = [] for child_price_min, child_price_max in price_splits[: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="price", split_label=_mobilede_price_label(child_price_min, child_price_max), split_params={"p": _mobilede_range_value(child_price_min, child_price_max)}, price_range=(child_price_min, child_price_max), ) ) price_kind = "recent_price" if year_min is not None and year_min >= 2023 else "price" _mobilede_append_overflow_candidate_group(candidate_groups, price_kind, child_segments) year_span = None if year_min is not None and year_max is not None: year_span = int(year_max) - int(year_min) allow_year_split = not ( previous_split_kind == "year" and depth >= MOBILEDE_OVERFLOW_YEAR_DEEP_SPLIT_DEPTH ) if allow_year_split: # Не уходим в слишком узкие year-ветки, если уже есть более полезная разбивка. if year_span is not None and year_span < max(4, MOBILEDE_OVERFLOW_MIN_YEAR_SPLIT_SPAN): allow_year_split = False elif price_splits or mileage_splits: allow_year_split = False year_splits = _mobilede_split_year_ranges_for_overflow(year_min, year_max) if allow_year_split else [] if year_splits: child_segments = [] for child_year_min, child_year_max in year_splits[: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="year", split_label=_mobilede_year_label(child_year_min, child_year_max), split_params={"fr": _mobilede_range_value(child_year_min, child_year_max)}, year_range=(child_year_min, child_year_max), ) ) _mobilede_append_overflow_candidate_group(candidate_groups, "year", child_segments) if mileage_splits: child_segments = [] for child_mileage_min, child_mileage_max in mileage_splits[: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(child_mileage_min, child_mileage_max), split_params={"ml": _mobilede_range_value(child_mileage_min, child_mileage_max)}, mileage_range=(child_mileage_min, child_mileage_max), ) ) _mobilede_append_overflow_candidate_group(candidate_groups, "mileage", child_segments) candidate_groups.sort(key=lambda item: {"recent_price": 0, "price": 1, "year": 2, "mileage": 3, "root_mileage": 4}.get(item[0], 9)) return candidate_groups def _mobilede_score_overflow_candidate_group( segment: dict[str, object], kind: str, children: list[dict[str, object]], ) -> tuple[int, int, int, int, int, int, int] | None: if not children: return None parent_total = _mobilede_segment_total_results(segment) page_cap = max(MOBILEDE_RESULTS_PER_PAGE, MOBILEDE_MAX_PAGE_NUMBER * MOBILEDE_RESULTS_PER_PAGE) overflow_children = 0 unresolved_children = 0 max_child_total = 0 sum_child_total = 0 tiny_children = 0 micro_children = 0 narrow_year_children = 0 target_band_misses = 0 distance_to_target = 0 probed = 0 depth = max(0, int(_mobilede_parse_optional_int(segment.get("overflow_depth")) or 0)) price_min = _mobilede_parse_optional_int(segment.get("price_min")) price_max = _mobilede_parse_optional_int(segment.get("price_max")) mileage_min = _mobilede_parse_optional_int(segment.get("mileage_min")) mileage_max = _mobilede_parse_optional_int(segment.get("mileage_max")) price_span = _mobilede_range_span(price_min, price_max) mileage_span = _mobilede_range_span(mileage_min, mileage_max) make_id = str(segment.get("make_id") or "").strip() previous_split_kind = str(segment.get("overflow_split") or "").strip() target, lower_target, tiny_threshold = _mobilede_target_results_band() for child in children[:MOBILEDE_OVERFLOW_SPLIT_PROBE_CHILDREN]: child_total = _mobilede_probe_segment_total(child) if child_total is None: unresolved_children += 1 continue child["total_results"] = child_total probed += 1 sum_child_total += int(child_total) max_child_total = max(max_child_total, int(child_total)) if int(child_total) > page_cap: overflow_children += 1 if int(child_total) < tiny_threshold: tiny_children += 1 if int(child_total) < lower_target: micro_children += 1 if not (lower_target <= int(child_total) <= target): target_band_misses += 1 distance_to_target += abs(target - int(child_total)) child_year_min = _mobilede_parse_optional_int(child.get("year_min")) child_year_max = _mobilede_parse_optional_int(child.get("year_max")) if ( child_year_min is not None and child_year_max is not None and int(child_year_max) - int(child_year_min) <= 1 ): narrow_year_children += 1 if probed == 0: return None if parent_total is None or parent_total <= 0: coverage_gap = 0 else: coverage_gap = abs(int(parent_total) - int(sum_child_total)) split_bias = { "recent_price": 0, "price": 1, "year": 2, "mileage": 3, "root_mileage": 4, }.get(kind, 5) if kind == "year": split_bias += MOBILEDE_OVERFLOW_SCORE_YEAR_PENALTY split_bias += depth * MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY split_bias += narrow_year_children * MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY elif kind in {"mileage", "root_mileage"}: split_bias += MOBILEDE_OVERFLOW_SCORE_MILEAGE_PENALTY split_bias += depth * max(1, MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY // 2) if kind == "mileage" and previous_split_kind == "mileage": split_bias += MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY + MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY if price_span is not None and price_span <= 10000: split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY if price_span is not None and price_span <= 5000: split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY if mileage_span is not None and mileage_span <= 100000: split_bias += max(1, MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY // 2) if mileage_span is not None and mileage_span <= 50000: split_bias += MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY if mileage_span is not None and mileage_span <= 25000: split_bias += MOBILEDE_OVERFLOW_SCORE_DEPTH_PENALTY + MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY distance_to_target += micro_children * max(1, MOBILEDE_OVERFLOW_SCORE_MICRO_CHILD_PENALTY // 2) return ( overflow_children, tiny_children + micro_children + narrow_year_children, target_band_misses, distance_to_target, coverage_gap, max_child_total, split_bias + unresolved_children, ) def _mobilede_overflow_candidate_group_is_useful( segment: dict[str, object], kind: str, children: list[dict[str, object]], ) -> bool: if not children: return False target, lower_target, tiny_threshold = _mobilede_target_results_band() useful_floor = max(MOBILEDE_RESULTS_PER_PAGE, int(target * MOBILEDE_OVERFLOW_MIN_USEFUL_CHILD_RATIO)) child_totals = [ int(total) for total in ( _mobilede_parse_optional_int(child.get("total_results")) for child in children[:MOBILEDE_OVERFLOW_SPLIT_PROBE_CHILDREN] ) if total is not None ] if not child_totals: return True useful_children = sum(total >= useful_floor for total in child_totals) non_empty_children = sum(total > 0 for total in child_totals) micro_children = sum(total < lower_target for total in child_totals) tiny_children = sum(total < tiny_threshold for total in child_totals) max_child_total = max(child_totals) sum_child_total = sum(child_totals) parent_total = _mobilede_segment_total_results(segment) depth = max(0, int(_mobilede_parse_optional_int(segment.get("overflow_depth")) or 0)) previous_split_kind = str(segment.get("overflow_split") or "").strip() price_span = _mobilede_range_span( _mobilede_parse_optional_int(segment.get("price_min")), _mobilede_parse_optional_int(segment.get("price_max")), ) mileage_span = _mobilede_range_span( _mobilede_parse_optional_int(segment.get("mileage_min")), _mobilede_parse_optional_int(segment.get("mileage_max")), ) if parent_total is not None and int(parent_total) > target: if non_empty_children > 0 and sum_child_total >= int(parent_total * 0.5): return True if max_child_total >= lower_target: return True if tiny_children >= len(child_totals): return False if micro_children > MOBILEDE_OVERFLOW_MAX_MICRO_CHILDREN and useful_children == 0: return False if kind in {"mileage", "root_mileage"}: if previous_split_kind == "mileage" and useful_children == 0: return False if depth >= 2 and useful_children == 0: return False if price_span is not None and price_span <= 10000 and useful_children == 0: return False if mileage_span is not None and mileage_span <= 100000 and micro_children > 0 and useful_children == 0: return False if kind == "year" and useful_children == 0 and max_child_total < lower_target: return False return useful_children > 0 or max_child_total >= lower_target def _mobilede_build_overflow_child_segments( *, segment: dict[str, object], max_pages: int, probe_best_split: bool = False, ) -> list[dict[str, object]]: candidate_groups = _mobilede_build_overflow_candidate_groups(segment=segment, max_pages=max_pages) if not candidate_groups: return [] if not probe_best_split or not MOBILEDE_OVERFLOW_SMART_SPLIT_ENABLED or len(candidate_groups) == 1: return candidate_groups[0][1] best_kind, best_children = candidate_groups[0] best_score = _mobilede_score_overflow_candidate_group(segment, best_kind, best_children) for kind, children in candidate_groups[1:MOBILEDE_OVERFLOW_SPLIT_PROBE_CANDIDATES]: score = _mobilede_score_overflow_candidate_group(segment, kind, children) if score is None: continue if best_score is None or score < best_score: best_kind = kind best_children = children best_score = score if best_score is not None: logger.info( "mobile.de overflow split selected: segment=%s split=%s score=%s candidates=%s", _mobilede_short_segment_label(segment), best_kind, best_score, ", ".join(kind for kind, _children in candidate_groups[:MOBILEDE_OVERFLOW_SPLIT_PROBE_CANDIDATES]), ) if not _mobilede_overflow_candidate_group_is_useful(segment, best_kind, best_children): logger.info( "mobile.de overflow split rejected as low-yield: segment=%s split=%s child_totals=%s", _mobilede_short_segment_label(segment), best_kind, [ _mobilede_parse_optional_int(child.get("total_results")) for child in best_children[:MOBILEDE_OVERFLOW_SPLIT_PROBE_CHILDREN] ], ) return [] return best_children def _mobilede_probe_segment_total(segment: dict[str, object]) -> int | None: search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() if not search_url: return None return _mobilede_probe_total(search_url) def _mobilede_touch_planning_progress(stage: str = "runtime_segments_planning") -> None: """Refresh global progress while the expensive segment planner is probing mobile.de.""" try: redis_client = _get_redis() redis_client.set(GLOBAL_PROGRESS_TS_KEY, str(int(time.time())), ex=2 * 60 * 60) except Exception: logger.debug("Failed to touch mobile.de planning progress: stage=%s", stage, exc_info=True) def _mobilede_segment_needs_preplan_split(segment: dict[str, object], total_results: int | None) -> bool: if total_results is None: return False threshold = min(MOBILEDE_SEGMENT_TARGET_RESULTS, _mobilede_overflow_threshold(MOBILEDE_MAX_PAGE_NUMBER)) return int(total_results) > int(threshold * MOBILEDE_PREPLAN_SPLIT_THRESHOLD_RATIO) def _mobilede_finalize_preplanned_segment(segment: dict[str, object], total_results: int | None) -> dict[str, object]: item = dict(segment) item["start_page"] = 1 item["max_pages"] = _mobilede_segment_pages_for_total( total_results, int(item.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER), ) item["total_results"] = total_results if total_results is not None: label = str(item.get("label") or _mobilede_segment_key(item)) label = re.sub(r"\s*\|\s*total=\d+", "", label) item["label"] = f"{label} | total={total_results}" return item def _mobilede_preplan_segment_tree(segment: dict[str, object], probe_budget: dict[str, int]) -> list[dict[str, object]]: """Probe and split a segment before dispatching any sync task. The plan is intentionally hybrid: deterministic segments are built first, then only a small global probe budget is used to catch obviously dense ranges. This prevents a long recursive probe phase before cars start being processed. """ pending: list[dict[str, object]] = [dict(segment)] planned: list[dict[str, object]] = [] seen: set[str] = set() while pending: if probe_budget["used"] >= probe_budget["limit"]: logger.warning( "mobile.de preplan probe budget reached: segment=%s probes=%s limit=%s pending=%s planned=%s", _mobilede_short_segment_label(segment), probe_budget["used"], probe_budget["limit"], len(pending), len(planned), ) planned.extend(_mobilede_finalize_preplanned_segment(item, None) for item in pending) break if len(planned) + len(pending) >= MOBILEDE_PREPLAN_MAX_SEGMENTS: logger.warning( "mobile.de preplan segment limit reached: segment=%s limit=%s pending=%s planned=%s", _mobilede_short_segment_label(segment), MOBILEDE_PREPLAN_MAX_SEGMENTS, len(pending), len(planned), ) planned.extend(_mobilede_finalize_preplanned_segment(item, None) for item in pending) break current = pending.pop(0) fingerprint = _mobilede_segment_fingerprint(current) if fingerprint in seen: continue seen.add(fingerprint) total_results = _mobilede_probe_segment_total(current) probe_budget["used"] += 1 if _mobilede_should_skip_planned_segment(total_results): continue depth = max(0, int(_mobilede_parse_optional_int(current.get("overflow_depth")) or 0)) max_preplan_depth = min(MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH, MOBILEDE_PREPLAN_MAX_SPLIT_DEPTH) if _mobilede_segment_needs_preplan_split(current, total_results) and depth < max_preplan_depth: children = _mobilede_build_overflow_child_segments( segment=current, max_pages=int(current.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER), probe_best_split=True, ) new_children = [child for child in children if _mobilede_segment_fingerprint(child) not in seen] if new_children: logger.info( "mobile.de preplan split: segment=%s total=%s children=%s depth=%s", _mobilede_short_segment_label(current), total_results, len(new_children), depth, ) pending.extend(new_children) continue logger.warning( "mobile.de preplan could not split dense segment: segment=%s total=%s depth=%s max_depth=%s", _mobilede_short_segment_label(current), total_results, depth, max_preplan_depth, ) planned.append(_mobilede_finalize_preplanned_segment(current, total_results)) return planned def _mobilede_preplan_runtime_segments(segments: list[dict[str, object]]) -> list[dict[str, object]]: if not MOBILEDE_PREPLAN_SEGMENT_PROBES: return segments planned: list[dict[str, object]] = [] probe_budget = {"used": 0, "limit": MOBILEDE_PREPLAN_MAX_PROBES} for index, segment in enumerate(segments): if len(planned) >= MOBILEDE_PREPLAN_MAX_SEGMENTS: remaining = segments[index:] logger.warning( "mobile.de global preplan segment limit reached: limit=%s planned=%s remaining=%s", MOBILEDE_PREPLAN_MAX_SEGMENTS, len(planned), len(remaining), ) planned.extend(_mobilede_finalize_preplanned_segment(item, None) for item in remaining) break if probe_budget["used"] >= probe_budget["limit"]: remaining = segments[index:] logger.warning( "mobile.de preplan switches to no-probe mode: probes=%s limit=%s planned=%s remaining=%s", probe_budget["used"], probe_budget["limit"], len(planned), len(remaining), ) planned.extend(_mobilede_finalize_preplanned_segment(item, None) for item in remaining) break planned.extend(_mobilede_preplan_segment_tree(segment, probe_budget)) if len(planned) > MOBILEDE_PREPLAN_MAX_SEGMENTS: logger.warning( "mobile.de global preplan segment list truncated: limit=%s planned_before_truncate=%s", MOBILEDE_PREPLAN_MAX_SEGMENTS, len(planned), ) planned = planned[:MOBILEDE_PREPLAN_MAX_SEGMENTS] break logger.info( "mobile.de preplanned final segment list: input=%s final=%s probes_used=%s probe_limit=%s", len(segments), len(planned), probe_budget["used"], probe_budget["limit"], ) return planned 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 known_total = _mobilede_segment_total_results(segment) page_cap = max(MOBILEDE_RESULTS_PER_PAGE, normalized_max_pages * MOBILEDE_RESULTS_PER_PAGE) if known_total is not None and known_total <= page_cap: logger.info( "mobile.de overflow split skipped: segment=%s observed=%s threshold=%s total=%s cap=%s reason=known_total_within_cap", _mobilede_short_segment_label(segment), observed, threshold, known_total, page_cap, ) return 0 if known_total is None: known_total = _mobilede_probe_segment_total(segment) if known_total is not None and known_total <= page_cap: logger.info( "mobile.de overflow split skipped: segment=%s observed=%s threshold=%s total=%s cap=%s reason=probed_total_within_cap", _mobilede_short_segment_label(segment), observed, threshold, known_total, page_cap, ) return 0 child_segments = _mobilede_build_overflow_child_segments( segment=segment, max_pages=normalized_max_pages, probe_best_split=True, ) if not child_segments: logger.warning( "mobile.de overflow split skipped: segment=%s observed=%s threshold=%s depth=%s reason=no_child_segments", _mobilede_short_segment_label(segment), observed, threshold, int(_mobilede_parse_optional_int((segment or {}).get("overflow_depth")) or 0), ) 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 = _mobilede_prune_overflow_parent_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, ) _mobilede_save_learned_runtime_segments( _mobilede_source_segments_from_settings(settings), updated_segments, ) 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 parent_key=%s observed=%s threshold=%s added=%s total_segments=%s", _mobilede_short_segment_label(segment), _mobilede_short_segment_ref(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_get_overflow_child_segments( redis_client: Redis, settings: Settings, *, parent_fingerprint: str, ) -> list[tuple[int, dict[str, object]]]: cached_segments = _get_cached_mobilede_runtime_segments(redis_client) if cached_segments is None: cached_segments = _build_mobilede_runtime_segments(settings) children: list[tuple[int, dict[str, object]]] = [] for index, item in enumerate(cached_segments): if not isinstance(item, dict): continue if str(item.get("overflow_parent") or "").strip() != parent_fingerprint: continue children.append((index, dict(item))) return children def _queue_mobilede_overflow_child_segments( redis_client: Redis, settings: Settings, *, parent_segment: dict[str, object] | None, lane: str, delay_seconds: float, use_cursor: bool, only_new: bool | None, bootstrap_run: bool, refresh_cycle_id: str | None, ) -> int: if not parent_segment: return 0 if bootstrap_run and only_new is not True and _mobilede_skip_late_overflow_children_during_bootstrap(): logger.info( "mobile.de late overflow children not queued during bootstrap: parent=%s reason=preplan_first_pass", _mobilede_short_segment_label(parent_segment), ) return 0 parent_fingerprint = _mobilede_segment_fingerprint(parent_segment) children = _mobilede_get_overflow_child_segments( redis_client, settings, parent_fingerprint=parent_fingerprint, ) if not children: return 0 scope = refresh_cycle_id or ("bootstrap" if bootstrap_run else "runtime") queue_key = MOBILEDE_OVERFLOW_CHILDREN_QUEUED_KEY_FMT.format( scope=scope, parent_key=parent_fingerprint, ) ttl = max(3600, int(MOBILEDE_REFRESH_CYCLE_TTL_SECONDS)) queued = 0 for segment_index, child in children: child_fingerprint = _mobilede_segment_fingerprint(child) if int(redis_client.sadd(queue_key, child_fingerprint)) != 1: continue redis_client.expire(queue_key, ttl) if bootstrap_run and only_new is not True: _mobilede_try_mark_bootstrap_segment_dispatched(redis_client, child, ttl_seconds=ttl) mobilede_sync_search_task.apply_async( kwargs={ "start_page": int(child.get("start_page") or 1), "max_pages": int(child.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER), "lane": lane, "only_new": only_new, "search_url": str(child.get("search_url") or child.get("listing_url") or "").strip() or None, "make_id": str(child.get("make_id") or "").strip() or None, "model_id": str(child.get("model_id") or "").strip() or None, "price_min": str(child.get("price_min") or "").strip() or None, "price_max": str(child.get("price_max") or "").strip() or None, "year_min": str(child.get("year_min") or "").strip() or None, "year_max": str(child.get("year_max") or "").strip() or None, "mileage_min": str(child.get("mileage_min") or "").strip() or None, "mileage_max": str(child.get("mileage_max") or "").strip() or None, "delay_seconds": delay_seconds, "use_cursor": use_cursor, "continuous": True, "segment": child, "segment_index": segment_index, "runtime_rotation": False, "bootstrap_run": bootstrap_run, "refresh_cycle_id": refresh_cycle_id, }, queue=MOBILEDE_SYNC_QUEUE, countdown=0, ) queued += 1 if queued: logger.info( "mobile.de overflow children queued immediately: parent=%s parent_key=%s queued=%s scope=%s", _mobilede_short_segment_label(parent_segment), _mobilede_short_segment_ref(parent_segment), queued, scope, ) return queued def _mobilede_probe_total(search_url: str, **params: str | int | None) -> int | None: try: client = MobileDeClient.for_worker(delay_seconds=0) probe_timeout = max(5, int(os.getenv("MOBILEDE_PLAN_PROBE_TIMEOUT_SECONDS", "12"))) page = client.fetch_search_page( page_number=1, search_url=search_url, timeout=probe_timeout, max_retries=0, **params, ) return int(page.total_results or 0) except Exception as exc: logger.warning("mobile.de segment probe failed: params=%s error=%s", params, exc) return None 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, 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) def _mobilede_make_expanded_segment( base_segment: dict[str, object], *, search_url: str, base_label: str, label_parts: list[str], params: dict[str, str | int | None], total_results: int | None, fallback_max_pages: int, ) -> dict[str, object]: item = dict(base_segment) item["search_url"] = _mobilede_make_segment_url(search_url, **params) item["listing_url"] = item["search_url"] item["start_page"] = 1 item["max_pages"] = _mobilede_segment_pages_for_total(total_results, fallback_max_pages) item["total_results"] = total_results total_label = str(total_results) if total_results is not None else "unknown" item["label"] = f"{base_label} | {' | '.join(label_parts)} | total={total_label}" 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_make_probe_planned_segment( base_segment: dict[str, object], *, search_url: str, base_label: str, label_parts: list[str], params: dict[str, str | int | None], total_results: int | None, fallback_max_pages: int, price_range: tuple[int, 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]: item = _mobilede_make_expanded_segment( base_segment, search_url=search_url, base_label=base_label, label_parts=label_parts, params=params, total_results=total_results, fallback_max_pages=fallback_max_pages, ) if price_range is not None: _mobilede_set_segment_range_fields(item, price_min=price_range[0], price_max=price_range[1]) if year_range is not None: price_min = int(item.get("price_min") or 1) price_max_raw = str(item.get("price_max") or "").strip() _mobilede_set_segment_range_fields( item, price_min=price_min, price_max=int(price_max_raw) if price_max_raw else None, year_range=year_range, ) if mileage_range is not None: price_min = int(item.get("price_min") or 1) price_max_raw = str(item.get("price_max") or "").strip() year_min_raw = str(item.get("year_min") or "").strip() year_max_raw = str(item.get("year_max") or "").strip() _mobilede_set_segment_range_fields( item, price_min=price_min, price_max=int(price_max_raw) if price_max_raw else None, year_range=(int(year_min_raw) if year_min_raw else None, int(year_max_raw) if year_max_raw else None), mileage_range=mileage_range, ) return item 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["source_search_url"] = str(segment.get("source_search_url") or 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_by_probe(segment: dict[str, object]) -> list[dict[str, object]] | None: if not MOBILEDE_PREPLAN_SEGMENT_PROBES or not _mobilede_filtered_url_uses_adaptive_plan(): return None search_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip() if not search_url: return None base_label = str(segment.get("label") or "mobile.de segmented URL") max_pages = int(segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER) planned: list[dict[str, object]] = [] probes_used = 0 probe_limit = max(1, int(MOBILEDE_PREPLAN_MAX_PROBES)) started_at = time.monotonic() max_seconds = int(MOBILEDE_ADAPTIVE_URL_MAX_SECONDS) def _probe(params: dict[str, str | int | None]) -> int | None: nonlocal probes_used if probes_used >= probe_limit: return None if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: return None probes_used += 1 if probes_used == 1 or probes_used % 25 == 0: _mobilede_touch_planning_progress("adaptive_url_planning") if probes_used % 50 == 0: logger.info( "mobile.de adaptive planning progress: base=%s probes=%s/%s planned=%s", _mobilede_short_segment_label(segment), probes_used, probe_limit, len(planned), ) return _mobilede_probe_total(search_url, **params) time_budget_reached = False for price_min, price_max in _mobilede_price_ranges_for_segment(segment): if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: time_budget_reached = True break price_params: dict[str, str | int | None] = {"p": _mobilede_range_value(price_min, price_max)} price_label = _mobilede_price_label(price_min, price_max) price_total = _probe(price_params) if _mobilede_should_skip_planned_segment(price_total): continue if price_total is None or price_total <= MOBILEDE_SEGMENT_TARGET_RESULTS: planned.append( _mobilede_make_probe_planned_segment( segment, search_url=search_url, base_label=base_label, label_parts=[price_label], params=price_params, total_results=price_total, fallback_max_pages=max_pages, price_range=(price_min, price_max), ) ) continue for year_min, year_max in _mobilede_year_ranges_for_segment_price(segment, price_min, price_max): if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: time_budget_reached = True break year_label = _mobilede_year_label(year_min, year_max) for refined_price_min, refined_price_max in _mobilede_price_subranges_for_hot_year(price_min, price_max, year_min, year_max): if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: time_budget_reached = True break refined_price_label = _mobilede_price_label(refined_price_min, refined_price_max) year_params = { "p": _mobilede_range_value(refined_price_min, refined_price_max), "fr": _mobilede_range_value(year_min, year_max), } year_total = _probe(year_params) if _mobilede_should_skip_planned_segment(year_total): continue if year_total is None or year_total <= MOBILEDE_SEGMENT_TARGET_RESULTS: planned.append( _mobilede_make_probe_planned_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, price_range=(refined_price_min, refined_price_max), year_range=(year_min, year_max), ) ) continue for mileage_min, mileage_max in _mobilede_mileage_ranges(): if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: time_budget_reached = True break 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 = _probe(mileage_params) if _mobilede_should_skip_planned_segment(mileage_total): continue planned.append( _mobilede_make_probe_planned_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, price_range=(refined_price_min, refined_price_max), year_range=(year_min, year_max), mileage_range=(mileage_min, mileage_max), ) ) if time_budget_reached: break if time_budget_reached: break if time_budget_reached: logger.warning( "mobile.de adaptive URL planning time budget reached: base=%s seconds=%s segments=%s probes=%s/%s", _mobilede_short_segment_label(segment), max_seconds, len(planned), probes_used, probe_limit, ) logger.info( "mobile.de adaptive URL segments planned: base=%s segments=%s probes=%s limit=%s", _mobilede_short_segment_label(segment), len(planned), probes_used, probe_limit, ) # Финальную dense-доразбивку делает _build_mobilede_runtime_segments. # Здесь только возвращаем probe-план с total_results, чтобы не терять хвосты # за mobile.de cap 50 страниц в сегментах >1000 результатов. return [_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in planned] def _mobilede_refine_dense_planned_segments(segments: list[dict[str, object]]) -> list[dict[str, object]]: threshold = min(MOBILEDE_SEGMENT_TARGET_RESULTS, _mobilede_overflow_threshold(MOBILEDE_MAX_PAGE_NUMBER)) pending: list[dict[str, object]] = [dict(item) for item in segments] refined: list[dict[str, object]] = [] probes_used = 0 probe_limit = max(1, int(MOBILEDE_PREPLAN_MAX_PROBES)) max_refine_depth = max(MOBILEDE_PREPLAN_MAX_SPLIT_DEPTH, MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH) started_at = time.monotonic() max_seconds = int(MOBILEDE_REFINE_ADAPTIVE_MAX_SECONDS) while pending: if max_seconds > 0 and time.monotonic() - started_at >= max_seconds: logger.warning( "mobile.de dense refine time budget reached: seconds=%s refined=%s pending=%s probes=%s/%s", max_seconds, len(refined), len(pending), probes_used, probe_limit, ) refined.extend(_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in pending) break current = pending.pop(0) total_results = current.get("total_results") if total_results is None and probes_used < probe_limit: total_results = _mobilede_probe_segment_total(current) current["total_results"] = total_results probes_used += 1 if probes_used == 1 or probes_used % 25 == 0: _mobilede_touch_planning_progress("dense_refine") if probes_used % 50 == 0: logger.info( "mobile.de dense refine progress: probes=%s/%s refined=%s pending=%s", probes_used, probe_limit, len(refined), len(pending), ) depth = max(0, int(_mobilede_parse_optional_int(current.get("overflow_depth")) or 0)) if ( total_results is not None and int(total_results) > threshold and depth < max_refine_depth and probes_used < probe_limit and len(refined) + len(pending) < MOBILEDE_PREPLAN_MAX_SEGMENTS ): children = _mobilede_build_overflow_child_segments( segment=current, max_pages=int(current.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER), probe_best_split=True, ) if children: for child in children: if probes_used >= probe_limit or len(refined) + len(pending) >= MOBILEDE_PREPLAN_MAX_SEGMENTS: pending.append(child) continue child_total = _mobilede_probe_segment_total(child) child["total_results"] = child_total probes_used += 1 if probes_used == 1 or probes_used % 25 == 0: _mobilede_touch_planning_progress("dense_refine") if probes_used % 50 == 0: logger.info( "mobile.de dense refine progress: probes=%s/%s refined=%s pending=%s", probes_used, probe_limit, len(refined), len(pending), ) if _mobilede_should_skip_planned_segment(child_total): continue pending.append(_mobilede_finalize_preplanned_segment(child, child_total)) continue if total_results is not None and int(total_results) > threshold: logger.warning( "mobile.de dense segment remains after refine: segment=%s total=%s depth=%s max_depth=%s probes=%s/%s", _mobilede_short_segment_label(current), total_results, depth, max_refine_depth, probes_used, probe_limit, ) refined.append(_mobilede_finalize_preplanned_segment(current, int(total_results) if total_results is not None else None)) logger.info( "mobile.de dense planned segments refined: input=%s output=%s probes=%s threshold=%s", len(segments), len(refined), probes_used, threshold, ) return refined def _mobilede_refine_dense_planned_segments_until_stable(segments: list[dict[str, object]]) -> list[dict[str, object]]: max_passes = max(1, int(os.getenv("MOBILEDE_REFINE_ADAPTIVE_MAX_PASSES", "2"))) refined = [dict(item) for item in segments] for pass_index in range(1, max_passes + 1): before = len(refined) refined = _mobilede_refine_dense_planned_segments(refined) dense_count = sum( 1 for item in refined if (_mobilede_segment_total_results(item) or 0) > min( MOBILEDE_SEGMENT_TARGET_RESULTS, _mobilede_overflow_threshold(MOBILEDE_MAX_PAGE_NUMBER), ) ) logger.info( "mobile.de dense refine pass complete: pass=%s/%s before=%s after=%s dense_left=%s", pass_index, max_passes, before, len(refined), dense_count, ) if dense_count <= 0 or len(refined) >= MOBILEDE_PREPLAN_MAX_SEGMENTS: break return refined def _mobilede_should_refine_adaptive_segments() -> bool: return os.getenv("MOBILEDE_REFINE_ADAPTIVE_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"} def _mobilede_try_keep_root_segment_unsplit( segment: dict[str, object], *, search_url: str, max_pages: int, ) -> list[dict[str, object]] | None: if not MOBILEDE_RUNTIME_BRAND_ROOT_PROBES_ENABLED: return None query_keys = {key for key, _ in parse_qsl(urlsplit(search_url).query, keep_blank_values=True)} if query_keys & {"p", "fr", "ml"}: return None total_results = _mobilede_probe_total(search_url) if _mobilede_should_skip_dynamic_segment(total_results): logger.info("mobile.de root segment skipped as empty: segment=%s", _mobilede_short_segment_label(segment)) return [] if total_results is None: return None threshold = min(MOBILEDE_SEGMENT_TARGET_RESULTS, _mobilede_overflow_threshold(max_pages)) if int(total_results) > threshold: logger.info( "mobile.de root segment requires split: segment=%s total=%s threshold=%s", _mobilede_short_segment_label(segment), total_results, threshold, ) return None item = _mobilede_finalize_preplanned_segment(segment, total_results) logger.info( "mobile.de root segment kept unsplit: segment=%s total=%s max_pages=%s", _mobilede_short_segment_label(item), total_results, item.get("max_pages"), ) return [item] def _expand_mobilede_search_url_segment( segment: dict[str, object], *, allow_adaptive_planning: bool = True, allow_probe_planning: bool = True, ) -> 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, allow_adaptive_planning=allow_adaptive_planning, allow_probe_planning=allow_probe_planning, ) ) return expanded # Готовый URL остаётся главным источником фильтров. Не режем его повторно # только если пользователь уже задал price/year/mileage в самой ссылке. # Остальные фильтры из URL (марка, тип кузова, топливо, страна и т.д.) # должны сохраниться, а поверх них можно добавить p/fr/ml для обхода 50-page cap. existing_range_keys = {"p", "fr", "ml"} query_keys = {key for key, _ in parse_qsl(urlsplit(search_url).query, keep_blank_values=True)} if query_keys & existing_range_keys: return [segment] max_pages = int(segment.get("max_pages") or MOBILEDE_MAX_PAGE_NUMBER) root_segment = _mobilede_try_keep_root_segment_unsplit( segment, search_url=search_url, max_pages=max_pages, ) if root_segment is not None: return root_segment if not allow_adaptive_planning: logger.info( "mobile.de URL segment fast-start enabled: using search_url as-is for %s", _mobilede_short_segment_label(segment), ) return [segment] if allow_probe_planning: probe_planned = _expand_mobilede_search_url_segment_by_probe(segment) if probe_planned is not None: return probe_planned expanded: list[dict[str, object]] = [] base_label = str(segment.get("label") or "mobile.de segmented URL") for price_min, price_max in _mobilede_price_ranges_for_segment(segment): params: dict[str, str | int | None] = {} if price_max is not None: params["p"] = f"{price_min}:{price_max}" else: params["p"] = f"{price_min}:" price_label = _mobilede_price_label(price_min, price_max) price_total = _mobilede_probe_total(search_url, **params) if MOBILEDE_DYNAMIC_SEGMENT_PROBES else None if _mobilede_should_skip_dynamic_segment(price_total): continue if price_total is not None and price_total <= MOBILEDE_SEGMENT_TARGET_RESULTS: item = _mobilede_make_expanded_segment( segment, search_url=search_url, base_label=base_label, label_parts=[price_label], params=params, total_results=price_total, fallback_max_pages=max_pages, ) _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_for_segment_price(segment, price_min, price_max): year_label = _mobilede_year_label(year_min, year_max) 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 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 def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, object]]: env_search_urls = settings.listing.filtered_search_urls # Готовые фильтрованные URL по умолчанию используют старое # детерминированное деление по цене/году: оно стартует сразу и его проще # держать стабильным. Адаптивное планирование на основе probe можно # включить явно, когда нужен более плотный pre-plan. fast_start_filtered_urls = ( bool(env_search_urls) and os.getenv("MOBILEDE_FILTERED_URL_FAST_START", "false").strip().lower() in {"1", "true", "yes", "on"} ) adaptive_filtered_urls = bool(env_search_urls) and _mobilede_filtered_url_uses_adaptive_plan() if env_search_urls: segments = _mobilede_source_segments_from_settings(settings) if any(str(item.get("runtime_brand") or "").strip() for item in segments): logger.info( "mobile.de runtime segments source: runtime brands via env filtered_search_urls templates=%s segments=%s", len(env_search_urls), len(segments), ) else: logger.info("mobile.de runtime segments source: env filtered_search_urls count=%s", len(env_search_urls)) else: segments = _mobilede_source_segments_from_settings(settings) learned_segments = _mobilede_load_learned_runtime_segments(segments) if learned_segments is not None: return _mobilede_interleave_segments_by_make(learned_segments) expanded: list[dict[str, object]] = [] expanded_from_adaptive_url = False for segment in segments: if segment.get("auto_segment") is False: expanded.append(segment) continue segment_expanded = _expand_mobilede_search_url_segment( segment, allow_adaptive_planning=not fast_start_filtered_urls, allow_probe_planning=not env_search_urls or adaptive_filtered_urls, ) expanded.extend(segment_expanded) if any(item.get("total_results") is not None for item in segment_expanded): expanded_from_adaptive_url = True if len(expanded) != len(segments): logger.info("mobile.de segments planned: input=%s total=%s", len(segments), len(expanded)) expanded = _mobilede_interleave_segments_by_make(expanded) if fast_start_filtered_urls: logger.info( "mobile.de fast-start runtime segments ready: input_urls=%s final=%s reason=filtered_search_urls", len(env_search_urls), len(expanded), ) return [_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in expanded] if expanded_from_adaptive_url: if _mobilede_should_refine_adaptive_segments(): refined = _mobilede_refine_dense_planned_segments_until_stable(expanded) logger.info( "mobile.de adaptive URL plan refined: before=%s after=%s reason=dense_segments", len(expanded), len(refined), ) return [_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in refined] logger.info( "mobile.de preplan skipped after adaptive URL planning: final=%s reason=fast_start", len(expanded), ) return [_mobilede_finalize_preplanned_segment(item, item.get("total_results")) for item in expanded] return _mobilede_preplan_runtime_segments(expanded)