mobile.de/iaai_scraper/worker/tasks.py

# Задачи Celery для синхронизации автомобилей и листинга IAAI.

import json
import logging
import os
import signal
import hashlib
from threading import Event, Thread
import time
import uuid
from urllib.parse import parse_qsl, urlsplit

from billiard.exceptions import SoftTimeLimitExceeded
from celery import shared_task
from redis import Redis
import requests

from ..core.config import Settings, build_fast_listing_segments_for_makes, build_listing_segments_for_makes, parse_listing_segments
from ..core.runtime_config import RuntimeConfig
from ..mobile_de import MobileDeClient, MobileDeScraper
from ..scraper import IAAIScraper
from ..storage.db import PersistenceService
from ..discovery import SitemapDiscoveryError, discover_vehicle_urls_from_sitemap_with_stats

logger = logging.getLogger("iaai_scraper.worker.tasks")

IAAI_SYNC_QUEUE = "iaai_sync"
MOBILEDE_SYNC_QUEUE = "mobilede_sync"
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_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_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_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_DYNAMIC_SEGMENT_PROBES = os.getenv("MOBILEDE_DYNAMIC_SEGMENT_PROBES", "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_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")))
MOBILEDE_ONLY_NEW_NEWEST_FIRST = os.getenv("MOBILEDE_ONLY_NEW_NEWEST_FIRST", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_INCREMENTAL_STRICT_FIRST_PASS = os.getenv("MOBILEDE_INCREMENTAL_STRICT_FIRST_PASS", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_ONLY_NEW_ZERO_INSERT_STREAK = max(1, int(os.getenv("MOBILEDE_ONLY_NEW_ZERO_INSERT_STREAK", "1")))
MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS = max(60, int(os.getenv("MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS", "3600")))
MOBILEDE_ONLY_NEW_HOT_ONLY = os.getenv("MOBILEDE_ONLY_NEW_HOT_ONLY", "true").strip().lower() in {"1", "true", "yes", "on"}
MOBILEDE_ONLY_NEW_HOT_TTL_SECONDS = max(300, int(os.getenv("MOBILEDE_ONLY_NEW_HOT_TTL_SECONDS", "10800")))
MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO = min(1.0, max(0.0, float(os.getenv("MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO", "0.95"))))
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_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_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}"

# Минимальная пауза между батчами (секунды) — не давит IAAI.
_INTER_BATCH_DELAY = max(float(os.getenv("IAAI_INTER_BATCH_DELAY_SECONDS", "0.3")), 0.0)
# Если доля failed в батче превышает порог — прерываем (IAAI блокирует).
_FAIL_RATE_THRESHOLD = min(max(float(os.getenv("IAAI_FAIL_RATE_THRESHOLD", "0.9")), 0.0), 1.0)

SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
SYNC_LISTING_CHECKPOINT_TTL_SECONDS = 7 * 24 * 60 * 60
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY = "iaai:state:sync_listing_bootstrap_failure_streak"
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT = 3
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS = 24 * 60 * 60
SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_KEY = "iaai:state:sync_listing_bootstrap_continuation_streak"
SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT = max(
    1,
    int(os.getenv("SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT", "6")),
)
SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_TTL_SECONDS = max(
    60,
    int(os.getenv("SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_TTL_SECONDS", str(6 * 60 * 60))),
)
HOURLY_FAILURE_STREAK_KEY = "iaai:state:hourly_failure_streak"
HOURLY_FAILURE_STREAK_LIMIT = 3
HOURLY_FAILURE_STREAK_TTL_SECONDS = 6 * 60 * 60  # сброс через 6 часов
SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
SYNC_LISTING_TASK_NAME = "iaai.sync_cars_feed"
SYNC_SEGMENT_LOCK_KEY_FMT = "iaai:locks:sync_segment:{idx}"
SYNC_SEGMENTS_PROGRESS_KEY = "iaai:state:sync_segments_progress"
SYNC_SEGMENTS_TOTAL_KEY = "iaai:state:sync_segments_total"
SYNC_SEGMENTS_PROGRESS_TTL_SECONDS = 24 * 60 * 60
TASK_PROGRESS_KEY_FMT = "iaai:state:task_progress:{task_id}"
GLOBAL_PROGRESS_TS_KEY = "iaai:state:last_progress_ts"
GLOBAL_DB_PROGRESS_TS_KEY = "iaai:state:last_db_progress_ts"
SITEMAP_HOURLY_LAST_COUNT_KEY = "iaai:state:sitemap_hourly_last_count"
SITEMAP_HOURLY_REFRESH_OFFSET_KEY = "iaai:state:sitemap_hourly_refresh_offset"
STALL_WATCHDOG_NAVIGATION_STAGES = {
    "listing_next_page_started",
    "listing_resume_progress",
}
STALL_WATCHDOG_LONG_RUNNING_STAGES = {
    "fast_listing_collected",
    "fast_detail_progress",
}
STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS = max(
    300,
    int(os.getenv("STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS", "900")),
)
STALL_WATCHDOG_DETAIL_GRACE_SECONDS = max(
    600,
    int(os.getenv("STALL_WATCHDOG_DETAIL_GRACE_SECONDS", "1200")),
)
DB_IDLE_RESTART_SECONDS = max(60, int(os.getenv("IAAI_DB_IDLE_RESTART_SECONDS", "3600")))
TERMINAL_PROGRESS_STAGES = {
    "segment_done",
    "segment_failed",
    "segment_task_completed",
    "segment_task_failed",
    "segment_task_soft_timeout",
    "sync_done",
    "failed",
}


def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
    last_exc: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            return func()
        except Exception as exc:
            last_exc = exc
            if attempt >= attempts:
                break
            delay = base_delay_s * (2 ** (attempt - 1))
            logger.warning(
                "Operation failed (attempt %d/%d): %s. Retrying in %.1fs",
                attempt,
                attempts,
                exc,
                delay,
            )
            time.sleep(delay)
    if last_exc is not None:
        raise last_exc


def _run_browser_job(func, *args, **kwargs):
    # С pool=solo Celery worker работает в одном процессе/потоке.
    # Playwright sync API использует greenlets, которые привязаны к потоку.
    # Запуск в отдельном потоке вызывает greenlet.error: cannot switch to a different thread.
    # Поэтому запускаем напрямую в текущем потоке.
    return func(*args, **kwargs)


def _sync_listing_lock_ttl_seconds() -> int:
    settings = Settings()
    soft = settings.celery.task_soft_time_limit
    hard = settings.celery.task_time_limit
    # Используем clamped hard limit (soft + 120), а не сырой task_time_limit,
    # чтобы lock не висел 11 дней при CELERY_TASK_TIME_LIMIT=999999.
    effective_hard = min(hard, soft + 120) if soft else hard
    return max(effective_hard + 120, 300)


def _sync_segment_lock_ttl_seconds() -> int:
    return 300


def _task_progress_key(task_id: str) -> str:
    return TASK_PROGRESS_KEY_FMT.format(task_id=task_id)


def _update_task_progress(
    redis_client: Redis,
    *,
    task_id: str,
    stage: str,
    ttl_seconds: int,
    **payload,
) -> None:
    try:
        now_ts = int(time.time())
        existing_task_started_ts: int | None = None
        try:
            existing_raw = redis_client.get(_task_progress_key(task_id))
            if existing_raw:
                existing_payload = json.loads(existing_raw)
                existing_task_started_ts = _safe_int(existing_payload.get("task_started_ts"))
        except Exception:
            existing_task_started_ts = None
        payload.setdefault("task_started_ts", existing_task_started_ts or now_ts)
        if stage != "fast_db_progress" and "last_db_progress_ts" not in payload:
            last_db_progress_ts = _safe_int(redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY))
            if last_db_progress_ts is not None:
                payload["last_db_progress_ts"] = last_db_progress_ts
        data = {
            "task_id": task_id,
            "stage": stage,
            "ts": now_ts,
            **payload,
        }
        ttl = max(60, int(ttl_seconds))
        pipe = redis_client.pipeline()
        pipe.set(
            _task_progress_key(task_id),
            json.dumps(data, ensure_ascii=False),
            ex=ttl,
        )
        # Глобальный маркер активности для внешнего guard-процесса.
        # Нужен, чтобы контейнер мог самовосстанавливаться при полном зависании воркера
        # (когда PID жив, но прогресс по задачам не двигается).
        pipe.set(GLOBAL_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
        if stage == "fast_db_progress":
            pipe.set(GLOBAL_DB_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
        pipe.execute()
    except Exception:
        logger.warning("Failed to update task progress for %s", task_id, exc_info=True)


def _clear_task_progress(redis_client: Redis, task_id: str) -> None:
    try:
        redis_client.delete(_task_progress_key(task_id))
    except Exception:
        logger.warning("Failed to clear task progress for %s", task_id, exc_info=True)


def _stall_timeout_for_progress(stage: str | None, default_timeout: int) -> int:
    if stage in STALL_WATCHDOG_NAVIGATION_STAGES:
        return max(int(default_timeout), STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS)
    if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
        return max(int(default_timeout), STALL_WATCHDOG_DETAIL_GRACE_SECONDS)
    return int(default_timeout)


def _hourly_sitemap_diff_sync(*, lane: str, limit: int | None, only_new: bool | None, progress_callback=None) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    existing_urls = persistence.get_all_active_origin_urls_for_lane("iaai:")

    new_urls = [url for url in discovered_urls if url not in existing_urls]
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    cars_upserted = 0
    cars_failed = 0
    protection_events = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    if new_urls:
        with IAAIScraper(settings) as scraper:
            if progress_callback:
                scraper.set_progress_callback(progress_callback)
            batch_size = settings.celery.batch_size
            for batch_start in range(0, len(new_urls), batch_size):
                batch_urls = new_urls[batch_start:batch_start + batch_size]
                batch_result = scraper.sync_batch(batch_urls, lane=lane)
                batch_ok = int(batch_result.get("cars_upserted", 0))
                batch_fail = int(batch_result.get("cars_failed", 0))
                batch_protection = int(batch_result.get("protection_events", 0))
                cars_upserted += batch_ok
                cars_failed += batch_fail
                protection_events += batch_protection
                images_upserted += int(batch_result.get("images_upserted", 0))
                failures.extend(batch_result.get("failures", []))
                # Fail-fast: если слишком много ошибок — IAAI блокирует, не тратим ресурсы.
                total_in_batch = batch_ok + batch_fail
                if total_in_batch > 0 and batch_fail / total_in_batch >= _FAIL_RATE_THRESHOLD:
                    logger.warning("Fail-fast: %d/%d failed in batch, stopping", batch_fail, total_in_batch)
                    break
                if batch_start + batch_size < len(new_urls):
                    time.sleep(_INTER_BATCH_DELAY)

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": len(discovered_urls) - len(new_urls),
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_diff",
        "discovered_urls": len(discovered_urls),
        "new_urls": len(new_urls),
        "sold_marked": sold_count,
        "protection_events": protection_events,
        "transport": discovery_result.stats.transport,
    }


def _hourly_sitemap_full_refresh_sync(*, lane: str, limit: int | None, only_new: bool | None, progress_callback=None) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    cars_upserted = 0
    cars_failed = 0
    protection_events = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    with IAAIScraper(settings) as scraper:
        if progress_callback:
            scraper.set_progress_callback(progress_callback)
        batch_size = settings.celery.batch_size
        for batch_start in range(0, len(discovered_urls), batch_size):
            batch_urls = discovered_urls[batch_start:batch_start + batch_size]
            batch_result = scraper.sync_batch(batch_urls, lane=lane)
            batch_ok = int(batch_result.get("cars_upserted", 0))
            batch_fail = int(batch_result.get("cars_failed", 0))
            batch_protection = int(batch_result.get("protection_events", 0))
            cars_upserted += batch_ok
            cars_failed += batch_fail
            protection_events += batch_protection
            images_upserted += int(batch_result.get("images_upserted", 0))
            failures.extend(batch_result.get("failures", []))
            total_in_batch = batch_ok + batch_fail
            if total_in_batch > 0 and batch_fail / total_in_batch >= _FAIL_RATE_THRESHOLD:
                logger.warning("Fail-fast: %d/%d failed in batch, stopping", batch_fail, total_in_batch)
                break
            if batch_start + batch_size < len(discovered_urls):
                time.sleep(_INTER_BATCH_DELAY)

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": 0,
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_full_refresh",
        "discovered_urls": len(discovered_urls),
        "new_urls": None,
        "sold_marked": sold_count,
        "protection_events": protection_events,
        "transport": discovery_result.stats.transport,
    }


def _hourly_sitemap_rolling_refresh_sync(
    *,
    redis_client: Redis,
    lane: str,
    limit: int | None,
    only_new: bool | None,
    progress_callback=None,
) -> dict[str, object]:
    del limit, only_new
    settings = Settings()
    persistence = _get_persistence()
    persistence.create_tables()

    discovery_result = discover_vehicle_urls_from_sitemap_with_stats(settings=settings)
    discovered_urls = discovery_result.vehicle_urls
    active_urls = set(discovered_urls)
    sold_count = 0
    if active_urls:
        sold_count = persistence.mark_sold_not_in_listing_by_urls(active_urls, lane="iaai")

    existing_urls = persistence.get_all_active_origin_urls_for_lane("iaai:")
    new_urls = [url for url in discovered_urls if url not in existing_urls]

    total_active = persistence.count_active_cars_for_lane("iaai:")
    batch_size = max(1, int(settings.discovery.hourly_refresh_batch_size))
    try:
        offset = int(redis_client.get(SITEMAP_HOURLY_REFRESH_OFFSET_KEY) or 0)
    except Exception:
        offset = 0

    refresh_urls = persistence.get_active_origin_urls_batch_for_refresh(
        prefix="iaai:",
        offset=offset,
        limit=batch_size,
    )
    if not refresh_urls and total_active > 0:
        offset = 0
        refresh_urls = persistence.get_active_origin_urls_batch_for_refresh(
            prefix="iaai:",
            offset=0,
            limit=batch_size,
        )

    next_offset = 0
    if total_active > 0:
        next_offset = offset + len(refresh_urls)
        if next_offset >= total_active:
            next_offset = 0

    try:
        redis_client.set(SITEMAP_HOURLY_REFRESH_OFFSET_KEY, str(next_offset))
    except Exception:
        logger.warning("Failed to persist hourly rolling refresh offset", exc_info=True)

    seen: set[str] = set()
    target_urls: list[str] = []
    for url in new_urls + refresh_urls:
        if url in seen:
            continue
        seen.add(url)
        target_urls.append(url)

    cars_upserted = 0
    cars_failed = 0
    protection_events = 0
    images_upserted = 0
    failures: list[dict[str, str]] = []

    if target_urls:
        with IAAIScraper(settings) as scraper:
            if progress_callback:
                scraper.set_progress_callback(progress_callback)
            worker_batch_size = settings.celery.batch_size
            for batch_start in range(0, len(target_urls), worker_batch_size):
                batch_urls = target_urls[batch_start:batch_start + worker_batch_size]
                batch_result = scraper.sync_batch(batch_urls, lane=lane)
                batch_ok = int(batch_result.get("cars_upserted", 0))
                batch_fail = int(batch_result.get("cars_failed", 0))
                batch_protection = int(batch_result.get("protection_events", 0))
                cars_upserted += batch_ok
                cars_failed += batch_fail
                protection_events += batch_protection
                images_upserted += int(batch_result.get("images_upserted", 0))
                failures.extend(batch_result.get("failures", []))
                total_in_batch = batch_ok + batch_fail
                if total_in_batch > 0 and batch_fail / total_in_batch >= _FAIL_RATE_THRESHOLD:
                    logger.warning("Fail-fast: %d/%d failed in batch, stopping", batch_fail, total_in_batch)
                    break
                if batch_start + worker_batch_size < len(target_urls):
                    time.sleep(_INTER_BATCH_DELAY)

    status = "success" if not failures else ("partial_success" if cars_upserted else "failed")
    return {
        "status": status,
        "run_id": None,
        "cars_upserted": cars_upserted,
        "cars_failed": cars_failed,
        "images_upserted": images_upserted,
        "skipped_existing": max(0, len(discovered_urls) - len(new_urls)),
        "elapsed_seconds": None,
        "failures": failures,
        "full_scan_completed": True,
        "hourly_mode": "sitemap_rolling_refresh",
        "discovered_urls": len(discovered_urls),
        "new_urls": len(new_urls),
        "refresh_urls": len(refresh_urls),
        "sold_marked": sold_count,
        "protection_events": protection_events,
        "transport": discovery_result.stats.transport,
        "refresh_offset": offset,
        "refresh_next_offset": next_offset,
        "active_total": total_active,
    }


def _start_stall_watchdog(
    redis_client: Redis,
    *,
    task_id: str,
    stall_timeout_seconds: int,
    lock_key: str | None = None,
    lock_owner: str | None = None,
    db_idle_restart_seconds: int | None = None,
) -> tuple[Event, Thread]:
    stop_event = Event()
    interval_seconds = max(5.0, min(30.0, stall_timeout_seconds / 3))

    def _watchdog() -> None:
        key = _task_progress_key(task_id)
        no_data_count = 0
        # Абсолютный дедлайн: если watchdog работает дольше 3× stall_timeout без прогресса — убиваем.
        watchdog_born = time.monotonic()
        absolute_deadline = stall_timeout_seconds * 3
        while not stop_event.wait(interval_seconds):
            db_idle_restart = False
            try:
                raw = redis_client.get(key)
                if not raw:
                    no_data_count += 1
                    elapsed_since_born = time.monotonic() - watchdog_born
                    if no_data_count % 5 == 0:
                        logger.warning(
                            "Stall watchdog: no progress data for task %s after %d checks (%.0fs)",
                            task_id, no_data_count, elapsed_since_born,
                        )
                    # Если прогресс-данных нет дольше stall_timeout — считаем задачу мёртвой.
                    if elapsed_since_born > stall_timeout_seconds:
                        logger.error(
                            "Task %s has no progress data for %.0fs (> %ds); treating as stalled",
                            task_id, elapsed_since_born, stall_timeout_seconds,
                        )
                    else:
                        continue
                else:
                    no_data_count = 0
                    data = json.loads(raw)
                    stage = data.get("stage")
                    last_ts = int(data.get("ts") or 0)
                    if not last_ts:
                        continue
                    db_idle_restart = bool(
                        db_idle_restart_seconds
                        and _should_restart_for_db_idle(data, db_idle_restart_seconds)
                    )
                    if db_idle_restart:
                        logger.error(
                            "Task %s has no DB writes for >%ss at segment=%s/%s stage=%s; full restart required",
                            task_id,
                            db_idle_restart_seconds,
                            data.get("segment_index"),
                            data.get("segments_total"),
                            stage,
                        )
                    else:
                        effective_stall_timeout = _stall_timeout_for_progress(stage, stall_timeout_seconds)
                        age = int(time.time()) - last_ts
                        if age < effective_stall_timeout:
                            watchdog_born = time.monotonic()  # reset absolute deadline on real progress
                            continue
                        logger.error(
                            "Task %s stalled for %ss at stage=%s payload=%s; cleaning up and restarting",
                            task_id,
                            age,
                            stage,
                            data,
                        )
            except Exception:
                logger.warning("Failed to inspect task progress for stall watchdog", exc_info=True)
                # Если Redis тоже не отвечает дольше дедлайна — убиваем.
                if time.monotonic() - watchdog_born > absolute_deadline:
                    logger.error("Stall watchdog: Redis unreachable for %.0fs; forcing kill", time.monotonic() - watchdog_born)
                else:
                    continue

            if db_idle_restart:
                _restart_bootstrap_from_first_segment(
                    redis_client,
                    reason=f"no DB writes for >{db_idle_restart_seconds}s",
                )

            # ── Pre-SIGTERM cleanup: release lock so next task can run ──
            if lock_key and lock_owner:
                try:
                    _release_lock_if_owner(redis_client, lock_key, lock_owner)
                    logger.info("Stall watchdog: released lock %s before SIGTERM", lock_key)
                except Exception:
                    # Force-delete if owner check fails (process is dying anyway)
                    try:
                        redis_client.delete(lock_key)
                        logger.info("Stall watchdog: force-deleted lock %s", lock_key)
                    except Exception:
                        logger.warning("Stall watchdog: failed to release lock %s", lock_key, exc_info=True)

            # ── Queue a followup task so parsing resumes after restart ──
            try:
                followup_ttl = max(180, int(stall_timeout_seconds) + 300)
                if _try_set_followup_pending(redis_client, ttl_seconds=followup_ttl):
                    from ..worker.celery_app import celery_app
                    celery_app.send_task(
                        SYNC_LISTING_TASK_NAME,
                        kwargs={},
                        queue=IAAI_SYNC_QUEUE,
                        countdown=15,
                        expires=followup_ttl,
                    )
                    logger.info("Stall watchdog: queued followup sync_listing_task after stall")
                else:
                    logger.info("Stall watchdog: followup already pending, skip duplicate enqueue")
            except Exception:
                try:
                    _clear_followup_pending(redis_client)
                except Exception:
                    pass
                logger.warning("Stall watchdog: failed to queue followup task", exc_info=True)

            # SIGTERM даёт процессу время на cleanup (закрыть DB, browser).
            # Celery перехватит SIGTERM и поднимет Terminated / warm shutdown.
            try:
                os.kill(os.getpid(), signal.SIGTERM)
            except OSError:
                pass
            # Даём 30 секунд на graceful shutdown, потом SIGKILL как последний resort.
            stop_event.wait(30)
            if not stop_event.is_set():
                logger.error("Task %s did not stop after SIGTERM; forcing SIGKILL", task_id)
                os.kill(os.getpid(), signal.SIGKILL)

    thread = Thread(target=_watchdog, name=f"task-stall-watchdog-{task_id[:8]}", daemon=True)
    thread.start()
    return stop_event, thread


def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) -> bool:
    stage = str(progress.get("stage") or "")
    if stage in TERMINAL_PROGRESS_STAGES:
        return False
    if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
        timeout = _stall_timeout_for_progress(stage, db_idle_restart_seconds)
        progress_ts = _safe_int(progress.get("ts")) or 0
        return progress_ts > 0 and int(time.time()) - progress_ts >= timeout

    segments_total = _safe_int(progress.get("segments_total"))
    segment_index = _safe_int(progress.get("segment_index"))
    if segments_total is None or segment_index is None:
        return False
    if segments_total <= 0 or segment_index >= segments_total - 1:
        return False

    now_ts = int(time.time())
    progress_ts = _safe_int(progress.get("ts")) or 0
    if progress_ts <= 0:
        return False

    db_progress_ts = _safe_int(progress.get("last_db_progress_ts"))
    if db_progress_ts is None:
        db_progress_ts = _read_global_db_progress_ts()
    task_started_ts = _safe_int(progress.get("task_started_ts")) or progress_ts
    last_db_or_start_ts = max(db_progress_ts or 0, task_started_ts)
    return now_ts - last_db_or_start_ts >= int(db_idle_restart_seconds)


def _safe_int(value) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _mobilede_should_skip_dynamic_segment(total_results: int | None) -> bool:
    return MOBILEDE_DYNAMIC_SEGMENT_PROBES and MOBILEDE_SKIP_EMPTY_DYNAMIC_SEGMENTS and total_results == 0


def _read_global_db_progress_ts() -> int | None:
    try:
        redis_client = _get_redis()
        raw = redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY)
        return _safe_int(raw)
    except Exception:
        return None


def _mobilede_segment_key(segment: dict[str, object] | None) -> str:
    if not segment:
        return "all"
    listing_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip()
    if listing_url:
        digest = hashlib.sha1(listing_url.encode("utf-8")).hexdigest()[:16]
        return f"url:{digest}"
    make_id = str(segment.get("make_id") or segment.get("makeId") or segment.get("make") or "all").strip()
    model_id = str(segment.get("model_id") or segment.get("modelId") or segment.get("model") or "all").strip()
    return f"{make_id}:{model_id}".replace(" ", "_")


def _mobilede_segment_fingerprint(segment: dict[str, object] | None) -> str:
    if not segment:
        return "all"
    payload = json.dumps(segment, ensure_ascii=False, sort_keys=True, default=str)
    return hashlib.sha1(payload.encode("utf-8")).hexdigest()[:16]


def _mobilede_cursor_key(segment: dict[str, object] | None) -> str:
    if not segment:
        return MOBILEDE_SEARCH_CURSOR_KEY
    return MOBILEDE_SEGMENT_CURSOR_KEY_FMT.format(segment_key=_mobilede_segment_fingerprint(segment))


def _mobilede_progress_page_counter_key(segment: dict[str, object] | None) -> str:
    return MOBILEDE_PROGRESS_PAGE_COUNTER_KEY_FMT.format(segment_key=_mobilede_segment_fingerprint(segment))


def _mobilede_segment_zero_insert_streak_key(segment: dict[str, object] | None) -> str:
    return f"mobilede:state:segment_zero_insert_streak:{_mobilede_segment_fingerprint(segment)}"


def _mobilede_segment_cooldown_key(segment: dict[str, object] | None) -> str:
    return f"mobilede:state:segment_cooldown:{_mobilede_segment_fingerprint(segment)}"


def _mobilede_segment_hot_key(segment: dict[str, object] | None) -> str:
    return f"mobilede:state:segment_hot:{_mobilede_segment_fingerprint(segment)}"


def _mobilede_segment_in_cooldown(redis_client: Redis, segment: dict[str, object] | None) -> bool:
    if not segment:
        return False
    return bool(redis_client.ttl(_mobilede_segment_cooldown_key(segment)) > 0)


def _mobilede_segment_is_hot(redis_client: Redis, segment: dict[str, object] | None) -> bool:
    if not segment:
        return False
    return bool(redis_client.ttl(_mobilede_segment_hot_key(segment)) > 0)


def _mobilede_has_hot_segments(redis_client: Redis, segments: list[dict[str, object]]) -> bool:
    for segment in segments:
        if _mobilede_segment_is_hot(redis_client, segment):
            return True
    return False


def _mobilede_update_segment_freshness_state(
    redis_client: Redis,
    *,
    segment: dict[str, object] | None,
    only_new: bool | None,
    inserted: int,
    listings: int,
) -> None:
    if not segment or only_new is not True:
        return
    streak_key = _mobilede_segment_zero_insert_streak_key(segment)
    cooldown_key = _mobilede_segment_cooldown_key(segment)
    hot_key = _mobilede_segment_hot_key(segment)
    listings_count = max(0, int(listings))
    if inserted > 0:
        ratio = (float(inserted) / float(listings_count)) if listings_count > 0 else 0.0
        if ratio >= MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO:
            redis_client.delete(streak_key)
            redis_client.delete(cooldown_key)
            redis_client.set(hot_key, "1", ex=MOBILEDE_ONLY_NEW_HOT_TTL_SECONDS)
            return
        # Есть новые, но доля слишком низкая: сегмент считается холодным для only_new режима.
        redis_client.delete(hot_key)
        redis_client.set(cooldown_key, "1", ex=MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS)
        logger.info(
            "mobile.de segment marked cold by ratio: segment=%s inserted=%s listings=%s ratio=%.3f threshold=%.3f cooldown=%ss",
            _mobilede_segment_label(segment),
            inserted,
            listings_count,
            ratio,
            MOBILEDE_ONLY_NEW_MIN_INSERT_RATIO,
            MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS,
        )
        return
    streak = int(redis_client.incr(streak_key))
    redis_client.expire(streak_key, 24 * 60 * 60)
    if streak >= MOBILEDE_ONLY_NEW_ZERO_INSERT_STREAK:
        redis_client.set(cooldown_key, "1", ex=MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS)
        logger.info(
            "mobile.de segment cooldown enabled: segment=%s streak=%s cooldown=%ss",
            _mobilede_segment_label(segment),
            streak,
            MOBILEDE_ONLY_NEW_COOLDOWN_SECONDS,
        )


def _mobilede_segment_label(segment: dict[str, object] | None) -> str:
    if not segment:
        return "all"
    label = str(segment.get("label") or "").strip()
    return label or _mobilede_segment_key(segment)


def _find_mobilede_runtime_segment(
    settings: Settings,
    *,
    search_url: str | None = None,
    make_id: str | None,
    model_id: str | None,
) -> dict[str, object] | None:
    target_search_url = str(search_url or "").strip()
    target_make_id = str(make_id or "").strip()
    target_model_id = str(model_id or "").strip()
    if not target_search_url and not target_make_id and not target_model_id:
        return None
    for candidate in _build_mobilede_runtime_segments(settings):
        candidate_search_url = str(candidate.get("search_url") or candidate.get("listing_url") or "").strip()
        if target_search_url and candidate_search_url == target_search_url:
            return candidate
        candidate_make_id = str(candidate.get("make_id") or "").strip()
        candidate_model_id = str(candidate.get("model_id") or "").strip()
        if candidate_make_id == target_make_id and candidate_model_id == target_model_id:
            return candidate
    return None


def _mobilede_segment_make(segment: dict[str, object] | None, fallback: str | None = None) -> str:
    if segment:
        listing_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip()
        if listing_url:
            return "filtered-url"
        value = str(segment.get("make") or "").strip()
        if value:
            return value
    return str(fallback or "all").strip() or "all"


def _mobilede_segment_model(segment: dict[str, object] | None, fallback: str | None = None) -> str:
    if segment:
        listing_url = str(segment.get("search_url") or segment.get("listing_url") or "").strip()
        if listing_url:
            return "filtered-url"
        value = str(segment.get("model") or "").strip()
        if value:
            return value
    return str(fallback or "all").strip() or "all"


def _mobilede_segment_uses_url(segment: dict[str, object] | None, search_url: str | None = None) -> bool:
    if search_url and str(search_url).strip():
        return True
    if not segment:
        return False
    return bool(str(segment.get("search_url") or segment.get("listing_url") or "").strip())


def _mobilede_filter_source(segment: dict[str, object] | None, search_url: str | None = None) -> str:
    return "search_url" if _mobilede_segment_uses_url(segment, search_url) else "params"


def _is_mobilede_transient_request_error(exc: Exception) -> bool:
    if isinstance(
        exc,
        (
            requests.exceptions.ConnectionError,
            requests.exceptions.Timeout,
            requests.exceptions.ProxyError,
            requests.exceptions.SSLError,
        ),
    ):
        return True
    text = str(exc).lower()
    return any(
        marker in text
        for marker in (
            "nameresolutionerror",
            "temporary failure in name resolution",
            "max retries exceeded",
            "connection refused",
            "read timed out",
            "connect timeout",
        )
    )


def _mobilede_task_result_summary(
    *,
    result: dict[str, object],
    segment: dict[str, object] | None,
    start_page: int,
    end_page: int,
    make_name: str,
    model_name: str,
) -> dict[str, object]:
    upsert = result.get("upsert") if isinstance(result.get("upsert"), dict) else {}
    return {
        "status": "success",
        "source": "mobile.de",
        "run_id": result.get("run_id"),
        "segment": _mobilede_segment_label(segment),
        "make": make_name,
        "model": model_name,
        "pages": {
            "start": start_page,
            "end": end_page,
            "count": end_page - start_page + 1,
        },
        "listing_count": int(result.get("listing_count", 0) or 0),
        "unique_listing_count": int(result.get("unique_listing_count", 0) or 0),
        "upsert": {
            "inserted": int(upsert.get("inserted", 0) or 0),
            "updated": int(upsert.get("updated", 0) or 0),
            "images_upserted": int(upsert.get("images_upserted", 0) or 0),
        },
    }


def _log_mobilede_progress_threshold(
    redis_client: Redis,
    *,
    task_id: str,
    segment: dict[str, object] | None,
    delta_pages: int,
    delta_cars: int,
    delta_images: int,
    start_page: int,
    end_page: int,
) -> None:
    if delta_pages <= 0:
        return
    try:
        counter_key = _mobilede_progress_page_counter_key(segment)
        total_pages = int(redis_client.incrby(counter_key, int(delta_pages)))
        redis_client.expire(counter_key, 7 * 24 * 60 * 60)
        previous_total = total_pages - int(delta_pages)
        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),
            total_pages,
            delta_pages,
            delta_cars,
            delta_images,
            start_page,
            end_page,
            task_id,
        )
    except Exception:
        logger.debug("Failed to update mobile.de aggregated progress", exc_info=True)


def _mobilede_url_query_value(search_url: str, key: str) -> str | None:
    for item_key, item_value in parse_qsl(urlsplit(search_url).query, keep_blank_values=True):
        if item_key == key and item_value != "":
            return item_value
    return None


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)


def _mobilede_apply_newest_sort_to_url(search_url: str | None) -> str | None:
    if not search_url:
        return search_url
    return MobileDeClient.build_search_url_from_existing(search_url, page_number=1, sb="doc", od="down")


def _mobilede_is_strict_first_pass_mode(redis_client: Redis, *, segment: dict[str, object] | None, only_new: bool | None) -> bool:
    return bool(
        MOBILEDE_INCREMENTAL_STRICT_FIRST_PASS
        and only_new is True
        and segment is not None
        and _mobilede_bootstrap_done(redis_client)
        and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP
    )


def _mobilede_reserve_incremental_cycle(redis_client: Redis, *, total_segments: int | None) -> tuple[str, bool]:
    total = max(1, int(total_segments or 1))
    cycle = str(redis_client.get(MOBILEDE_INCREMENTAL_CYCLE_KEY) or "1")
    seen = int(redis_client.incr(MOBILEDE_INCREMENTAL_CYCLE_SEEN_COUNT_KEY))
    if seen >= total:
        next_cycle = str(int(cycle) + 1)
        redis_client.set(MOBILEDE_INCREMENTAL_CYCLE_KEY, next_cycle)
        redis_client.set(MOBILEDE_INCREMENTAL_CYCLE_SEEN_COUNT_KEY, "0")
    return cycle, seen == 1


def _mobilede_cycle_cursor_key(base_cursor_key: str, cycle_id: str) -> str:
    return f"{base_cursor_key}:cycle:{cycle_id}"


def _mobilede_cycle_seen_set_key(cycle_id: str) -> str:
    return MOBILEDE_INCREMENTAL_CYCLE_SEEN_SET_KEY_FMT.format(cycle_id=cycle_id)


def _mobilede_try_mark_cycle_segment_seen(
    redis_client: Redis,
    *,
    cycle_id: str,
    segment: dict[str, object] | None,
    ttl_seconds: int = 24 * 60 * 60,
) -> bool:
    fingerprint = _mobilede_segment_fingerprint(segment)
    set_key = _mobilede_cycle_seen_set_key(cycle_id)
    added = int(redis_client.sadd(set_key, fingerprint))
    redis_client.expire(set_key, ttl_seconds)
    return added == 1


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:
        cycle_id = "1"
        redis_client.set(MOBILEDE_INCREMENTAL_CYCLE_KEY, cycle_id)
    return cycle_id


def _mobilede_reserve_strict_first_pass_segment(
    redis_client: Redis,
    settings: Settings,
    *,
    segment: dict[str, object] | None,
    segment_index: int | None,
) -> tuple[dict[str, object] | None, int | None, str]:
    segments = _get_cached_mobilede_runtime_segments(redis_client) or []
    total_segments = len(segments)
    cycle_id = _mobilede_current_cycle_id(redis_client)
    runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
    only_new = runtime_config.sync.only_new

    def _try_take_current() -> bool:
        return segment is not None and _mobilede_try_mark_cycle_segment_seen(
            redis_client,
            cycle_id=cycle_id,
            segment=segment,
        )

    if _try_take_current():
        return segment, segment_index, cycle_id

    for _ in range(max(1, total_segments)):
        reservation = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new)
        if reservation is None:
            break
        next_index, next_segment = reservation
        if _mobilede_try_mark_cycle_segment_seen(
            redis_client,
            cycle_id=cycle_id,
            segment=next_segment,
        ):
            return next_segment, next_index, cycle_id

    # Все сегменты цикла пройдены: начинаем новый цикл и берём первый доступный.
    cycle_id = str(int(cycle_id) + 1)
    redis_client.set(MOBILEDE_INCREMENTAL_CYCLE_KEY, cycle_id)
    redis_client.set(MOBILEDE_INCREMENTAL_CYCLE_SEEN_COUNT_KEY, "0")

    if segment is not None and _mobilede_try_mark_cycle_segment_seen(
        redis_client,
        cycle_id=cycle_id,
        segment=segment,
    ):
        return segment, segment_index, cycle_id

    for _ in range(max(1, total_segments)):
        reservation = _reserve_mobilede_runtime_segment(redis_client, settings, only_new=only_new)
        if reservation is None:
            break
        next_index, next_segment = reservation
        if _mobilede_try_mark_cycle_segment_seen(
            redis_client,
            cycle_id=cycle_id,
            segment=next_segment,
        ):
            return next_segment, next_index, cycle_id

    return segment, segment_index, cycle_id


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
    compact = os.getenv("MOBILEDE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
    if compact:
        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_year_ranges() -> list[tuple[int | None, int | None]]:
    compact = os.getenv("MOBILEDE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
    if compact:
        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_mileage_ranges() -> list[tuple[int | None, int | None]]:
    compact = os.getenv("MOBILEDE_COMPACT_SEGMENTS", "true").strip().lower() in {"1", "true", "yes", "on"}
    if compact:
        return [(None, 100000), (100001, 200000), (200001, None)]
    return [(None, 50000), (50001, 100000), (100001, 150000), (150001, 200000), (200001, None)]


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_probe_total(search_url: str, **params: str | int | None) -> int | None:
    try:
        client = MobileDeClient.for_worker(delay_seconds=0)
        page = client.fetch_search_page(page_number=1, search_url=search_url, **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, 100)
    pages = max(1, min(100, (int(total_results) + 19) // 20))
    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
    item["label"] = f"{base_label} | {' | '.join(label_parts)} | total~{total_results if total_results is not None else '?'}"
    return item


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]
    # Если пользователь уже задал узкий 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)}
    if query_keys & existing_range_keys:
        return [segment]

    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)
    for price_min, price_max in _mobilede_price_ranges():
        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,
            )
            item["price_min"] = str(price_min)
            item["price_max"] = str(price_max) if price_max is not None else None
            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)
            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):
                    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)
    return expanded


def _build_mobilede_runtime_segments(settings: Settings) -> list[dict[str, object]]:
    runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
    segments = [segment.to_task_kwargs() for segment in runtime_config.mobilede.segments]
    expanded: list[dict[str, object]] = []
    for segment in segments:
        if segment.get("auto_segment") is False:
            expanded.append(segment)
            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))
    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)
        if not cached_raw:
            return None
        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 cached mobile.de runtime segments", exc_info=True)
    return None


def _request_mobilede_runtime_segments_rebuild(redis_client: Redis) -> None:
    try:
        redis_client.set(MOBILEDE_RUNTIME_SEGMENTS_PENDING_KEY, "1", ex=15 * 60)
    except Exception:
        logger.debug("Failed to request mobile.de runtime segment rebuild", exc_info=True)


def _mobilede_bootstrap_done(redis_client: Redis) -> bool:
    if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED:
        return True
    return bool(redis_client.get(MOBILEDE_BOOTSTRAP_DONE_KEY))


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)
    return cursor_raw is not None and int(cursor_raw) >= segment_max_pages


def _mobilede_bootstrap_segment_done(redis_client: Redis, segment: dict[str, object] | None) -> bool:
    if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or not segment:
        return False
    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:
    if not MOBILEDE_BOOTSTRAP_FULL_SCAN_ENABLED or not segment:
        return
    segment_done_key = f"mobilede:state:bootstrap_segment_done:{_mobilede_segment_fingerprint(segment)}"
    try:
        if total_segments is not None:
            redis_client.set(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY, str(int(total_segments)))
        if redis_client.setnx(segment_done_key, "1"):
            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)
            logger.info(
                "mobile.de bootstrap segment completed: %s done=%s/%s",
                _mobilede_segment_label(segment),
                done,
                total,
            )
            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)
    except Exception:
        logger.debug("Failed to mark mobile.de bootstrap segment complete", exc_info=True)


def _reserve_next_mobilede_runtime_segment(
    redis_client: Redis,
    settings: Settings,
    *,
    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:
        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


def _enqueue_mobilede_runtime_segments(
    *,
    lane: str,
    delay_seconds: float,
    use_cursor: bool,
    continuous: bool,
) -> 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 []
    if not segments:
        return []
    try:
        redis_client.set(MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY, str(len(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))):
        reservation = _reserve_next_mobilede_runtime_segment(redis_client, settings, only_new=only_new)
        if reservation is None:
            break
        initial_reservations.append(reservation)
    for index, segment in initial_reservations:
        mobilede_sync_search_task.apply_async(
            kwargs={
                "start_page": int(segment.get("start_page") or 1),
                "max_pages": int(segment.get("max_pages") or 5),
                "lane": lane,
                "delay_seconds": delay_seconds,
                "use_cursor": use_cursor,
                "continuous": continuous,
                "segment": segment,
                "segment_index": index,
                "runtime_rotation": True,
            },
            queue=MOBILEDE_SYNC_QUEUE,
        )
    return segments


def _reserve_mobilede_runtime_segment(
    redis_client: Redis,
    settings: Settings,
    *,
    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:
        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, segment
        return None

    reserved = _try_reserve(require_hot=True, respect_cooldown=True)
    if reserved is not None:
        return reserved
    reserved = _try_reserve(require_hot=False, respect_cooldown=True)
    if reserved is not None:
        return reserved
    reserved = _try_reserve(require_hot=False, respect_cooldown=False)
    if reserved is not None:
        return reserved
    return None


def _reserve_mobilede_page_window(
    redis_client: Redis,
    *,
    requested_start_page: int,
    page_window_size: int,
    use_cursor: bool,
    cursor_key: str,
) -> tuple[int, int]:
    page_window_size = max(1, int(page_window_size))
    requested_start_page = max(1, int(requested_start_page))
    if not use_cursor:
        return requested_start_page, requested_start_page + page_window_size - 1
    redis_client.setnx(cursor_key, str(requested_start_page - 1))
    window_end = int(redis_client.incrby(cursor_key, page_window_size))
    window_start = max(1, window_end - page_window_size + 1)
    return window_start, window_end


def _reset_mobilede_page_cursor(redis_client: Redis, *, cursor_key: str, next_start_page: int = 1) -> None:
    redis_client.set(cursor_key, str(max(0, int(next_start_page) - 1)))


_persistence_instance: PersistenceService | None = None


def _get_persistence() -> PersistenceService:
    global _persistence_instance
    if _persistence_instance is not None:
        return _persistence_instance

    sett = Settings()
    persistence = PersistenceService(sett)

    def _ping_db() -> None:
        with persistence.engine.connect() as conn:
            conn.exec_driver_sql("SELECT 1")

    _retry_with_backoff(_ping_db, attempts=5, base_delay_s=1.0)
    _persistence_instance = persistence
    return persistence


_redis_instance: Redis | None = None


def _get_redis() -> Redis:
    global _redis_instance
    if _redis_instance is not None:
        try:
            _redis_instance.ping()
            return _redis_instance
        except Exception:
            _redis_instance = None

    sett = Settings()
    redis_client = Redis.from_url(
        sett.redis.url,
        decode_responses=True,
        socket_connect_timeout=sett.redis.socket_connect_timeout_seconds,
        socket_timeout=sett.redis.socket_timeout_seconds,
        health_check_interval=sett.redis.health_check_interval_seconds,
        retry_on_timeout=True,
    )

    def _ping_redis() -> None:
        redis_client.ping()

    _retry_with_backoff(_ping_redis, attempts=5, base_delay_s=1.0)
    _redis_instance = redis_client
    return redis_client


def _acquire_lock(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool:
    try:
        acquired = bool(redis_client.set(key, owner_token, nx=True, ex=ttl_seconds))
        if acquired:
            return True

        # Автовосстановление: если lock завис без TTL, считаем stale и пересоздаём.
        ttl = redis_client.ttl(key)
        if ttl is not None and ttl < 0:
            logger.warning("Detected stale lock without TTL, removing: %s", key)
            redis_client.delete(key)
            return bool(redis_client.set(key, owner_token, nx=True, ex=ttl_seconds))

        return False
    except Exception as exc:
        logger.warning("Failed to acquire lock %s", key, exc_info=True)
        return False


def _refresh_lock_if_owner(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool | None:
    try:
        refreshed = redis_client.eval(
            """
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
            end
            return 0
            """,
            1,
            key,
            owner_token,
            int(ttl_seconds),
        )
        return bool(refreshed)
    except Exception as exc:
        logger.warning("Failed to refresh lock %s", key, exc_info=True)
        return None


def _release_lock_if_owner(redis_client: Redis, key: str, owner_token: str) -> None:
    try:
        redis_client.eval(
            """
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            end
            return 0
            """,
            1,
            key,
            owner_token,
        )
    except Exception as exc:
        logger.warning("Failed to release lock %s", key, exc_info=True)


def _has_running_sync_listing_tasks(celery_app, *, exclude_task_id: str | None = None) -> bool:
    try:
        inspector = celery_app.control.inspect(timeout=1.0)
        snapshots = [
            inspector.active() or {},
            inspector.reserved() or {},
            inspector.scheduled() or {},
        ]
    except Exception:
        logger.warning("Failed to inspect Celery workers for running sync tasks", exc_info=True)
        return True

    for snapshot in snapshots:
        for entries in snapshot.values():
            for entry in entries or []:
                task_name = str(entry.get("name") or entry.get("request", {}).get("name") or "")
                if task_name != SYNC_LISTING_TASK_NAME:
                    continue
                # Исключаем текущую задачу — она не считается "другой запущенной"
                entry_id = str(entry.get("id") or entry.get("request", {}).get("id") or "")
                if exclude_task_id and entry_id == exclude_task_id:
                    continue
                return True
    return False


def _clear_orphan_sync_listing_lock(redis_client: Redis, celery_app, *, current_task_id: str | None = None) -> bool:
    try:
        owner_token = redis_client.get(SYNC_LISTING_LOCK_KEY)
        if not owner_token:
            return False
    except Exception:
        logger.warning("Failed to read sync listing lock before cleanup", exc_info=True)
        return False

    if _has_running_sync_listing_tasks(celery_app, exclude_task_id=current_task_id):
        logger.info("sync_listing lock preserved: active task still detected")
        return False

    try:
        ttl = redis_client.ttl(SYNC_LISTING_LOCK_KEY)
        redis_client.delete(SYNC_LISTING_LOCK_KEY)
        logger.warning(
            "Removed orphan sync_listing lock owner=%s ttl=%s after worker restart",
            owner_token,
            ttl,
        )
        return True
    except Exception:
        logger.warning("Failed to clear orphan sync listing lock", exc_info=True)
        return False


def _is_full_scan_done(redis_client: Redis) -> bool:
    try:
        value = redis_client.get(SYNC_FULL_SCAN_DONE_KEY)
    except Exception:
        logger.warning("Failed to read full scan state", exc_info=True)
        return False
    return str(value or "").strip() == "1"


def _set_full_scan_done(redis_client: Redis, done: bool) -> None:
    try:
        redis_client.set(SYNC_FULL_SCAN_DONE_KEY, "1" if done else "0")
    except Exception:
        logger.warning("Failed to persist full scan state", exc_info=True)


def _load_last_completed_segment(redis_client: Redis) -> int | None:
    # Segment-only checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
    try:
        raw = redis_client.get(SYNC_LISTING_CHECKPOINT_KEY)
    except Exception:
        logger.warning("Failed to read sync listing checkpoint", exc_info=True)
        return None
    if raw is None:
        return None
    try:
        value = int(str(raw).strip())
    except (TypeError, ValueError):
        logger.warning("Invalid sync listing checkpoint value %r; clearing", raw)
        _clear_sync_checkpoint(redis_client)
        return None
    if value < 0:
        _clear_sync_checkpoint(redis_client)
        return None
    return value


def _save_last_completed_segment(redis_client: Redis, segment_index: int) -> None:
    try:
        redis_client.set(
            SYNC_LISTING_CHECKPOINT_KEY,
            str(int(segment_index)),
            ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
        )
    except Exception:
        logger.warning("Failed to save sync listing checkpoint", exc_info=True)


def _restart_bootstrap_from_first_segment(redis_client: Redis, *, reason: str) -> None:
    """Clear bootstrap checkpoint so the next full scan starts from segment 1."""
    try:
        pipe = redis_client.pipeline()
        pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
        pipe.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
        pipe.delete(GLOBAL_PROGRESS_TS_KEY)
        pipe.delete(GLOBAL_DB_PROGRESS_TS_KEY)
        pipe.set(SYNC_FULL_SCAN_DONE_KEY, "0")
        pipe.execute()
        logger.error("Bootstrap restart requested from segment 1: %s", reason)
    except Exception:
        logger.warning("Failed to reset bootstrap checkpoint for DB-idle restart", exc_info=True)


def _clear_sync_checkpoint(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_CHECKPOINT_KEY)
    except Exception:
        logger.warning("Failed to clear sync listing checkpoint", exc_info=True)


def _try_set_followup_pending(redis_client: Redis, *, ttl_seconds: int) -> bool:
    try:
        return bool(redis_client.set(SYNC_LISTING_FOLLOWUP_PENDING_KEY, "1", nx=True, ex=max(60, int(ttl_seconds))))
    except Exception:
        logger.warning("Failed to set follow-up pending flag", exc_info=True)
        return True


def _clear_followup_pending(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
    except Exception:
        logger.warning("Failed to clear follow-up pending flag", exc_info=True)


def _bump_bootstrap_failure_streak(
    redis_client: Redis,
    *,
    reason: str,
) -> tuple[int, bool]:
    try:
        streak = int(redis_client.incr(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY))
        redis_client.expire(
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY,
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS,
        )
    except Exception:
        logger.warning("Failed to update bootstrap failure streak", exc_info=True)
        return 0, True

    should_enqueue = streak < SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT
    if should_enqueue:
        logger.warning(
            "Bootstrap failure streak %d/%d recorded (reason=%s)",
            streak,
            SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT,
            reason,
        )
    else:
        logger.error(
            "Bootstrap follow-up circuit breaker opened after %d consecutive failures (reason=%s)",
            streak,
            reason,
        )
    return streak, should_enqueue


def _clear_bootstrap_failure_streak(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY)
    except Exception:
        logger.warning("Failed to clear bootstrap failure streak", exc_info=True)


def _bump_bootstrap_continuation_streak(redis_client: Redis) -> tuple[int, bool]:
    """Счётчик подряд идущих bootstrap-followup запусков.

    Возвращает (streak, should_continue). Если should_continue=False,
    немедленные continuation блокируются до следующего beat-цикла.
    """
    try:
        streak = int(redis_client.incr(SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_KEY))
        redis_client.expire(
            SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_KEY,
            SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_TTL_SECONDS,
        )
    except Exception:
        logger.warning("Failed to update bootstrap continuation streak", exc_info=True)
        return 0, True

    should_continue = streak <= SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT
    if not should_continue:
        logger.error(
            "Bootstrap continuation breaker OPEN: streak=%d limit=%d",
            streak,
            SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT,
        )
    return streak, should_continue


def _clear_bootstrap_continuation_streak(redis_client: Redis) -> None:
    try:
        redis_client.delete(SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_KEY)
    except Exception:
        logger.warning("Failed to clear bootstrap continuation streak", exc_info=True)


def _bump_hourly_failure_streak(redis_client: Redis) -> int:
    """Инкрементирует счётчик ошибок hourly. Возвращает новое значение."""
    try:
        streak = int(redis_client.incr(HOURLY_FAILURE_STREAK_KEY))
        redis_client.expire(HOURLY_FAILURE_STREAK_KEY, HOURLY_FAILURE_STREAK_TTL_SECONDS)
        return streak
    except Exception:
        logger.warning("Failed to update hourly failure streak", exc_info=True)
        return 0


def _check_hourly_circuit_breaker(redis_client: Redis) -> tuple[int, bool]:
    """Проверяет открыт ли circuit breaker. Возвращает (streak, is_open)."""
    try:
        raw = redis_client.get(HOURLY_FAILURE_STREAK_KEY)
        streak = int(raw) if raw else 0
    except Exception:
        return 0, False
    is_open = streak >= HOURLY_FAILURE_STREAK_LIMIT
    if is_open:
        logger.error("Hourly circuit breaker OPEN (%d/%d failures) — skipping", streak, HOURLY_FAILURE_STREAK_LIMIT)
    return streak, is_open


def _clear_hourly_failure_streak(redis_client: Redis) -> None:
    try:
        redis_client.delete(HOURLY_FAILURE_STREAK_KEY)
    except Exception:
        pass


def _start_lock_heartbeat(
    redis_client: Redis,
    key: str,
    owner_token: str,
    ttl_seconds: int,
    stop_on_lost: bool = True,
) -> tuple[Event, Thread]:
    stop_event = Event()
    interval_seconds = max(5.0, min(30.0, ttl_seconds / 3))

    def _heartbeat() -> None:
        consecutive_failures = 0
        while not stop_event.wait(interval_seconds):
            refreshed = _refresh_lock_if_owner(redis_client, key, owner_token, ttl_seconds)
            if refreshed is False:
                logger.warning("Lost sync_listing lock ownership for %s", owner_token)
                if stop_on_lost:
                    return
                consecutive_failures += 1
                continue
            if refreshed is None:
                consecutive_failures += 1
                if consecutive_failures >= 5:
                    logger.error("Lock heartbeat failed %d times in a row for %s; giving up", consecutive_failures, owner_token)
                    return
            else:
                consecutive_failures = 0

    thread = Thread(target=_heartbeat, name="sync-listing-lock-heartbeat", daemon=True)
    thread.start()
    return stop_event, thread


def _reset_segments_progress(redis_client: Redis, total: int) -> None:
    try:
        pipe = redis_client.pipeline()
        pipe.delete(SYNC_SEGMENTS_PROGRESS_KEY)
        pipe.set(SYNC_SEGMENTS_TOTAL_KEY, str(int(total)), ex=SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
        pipe.execute()
    except Exception:
        logger.warning("Failed to reset segments progress", exc_info=True)


def _mark_segment_completed(redis_client: Redis, segment_index: int) -> tuple[int, int]:
    """Помечает сегмент завершённым. Возвращает (completed_count, total)."""
    try:
        pipe = redis_client.pipeline()
        pipe.sadd(SYNC_SEGMENTS_PROGRESS_KEY, str(int(segment_index)))
        pipe.expire(SYNC_SEGMENTS_PROGRESS_KEY, SYNC_SEGMENTS_PROGRESS_TTL_SECONDS)
        pipe.scard(SYNC_SEGMENTS_PROGRESS_KEY)
        pipe.get(SYNC_SEGMENTS_TOTAL_KEY)
        results = pipe.execute()
        completed = int(results[2] or 0)
        total = int(results[3] or 0) if results[3] else 0
        return completed, total
    except Exception:
        logger.warning("Failed to mark segment %d completed", segment_index, exc_info=True)
        return 0, 0


def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | None]]:
    """Возвращает сегменты листинга с учётом runtime_config.

    При IAAI_LISTING_SEGMENTS=runtime сегменты строятся из filters.brands / filters.include.brands.
    Остальные значения IAAI_LISTING_SEGMENTS сохраняют прежнее поведение: auto или JSON.
    """
    filtered_urls = settings.listing.filtered_search_urls
    if len(filtered_urls) > 1:
        return [
            {"make": None, "year_min": None, "year_max": None, "listing_url": url}
            for url in filtered_urls
        ]
    if len(filtered_urls) == 1:
        return []

    raw_segments = settings.listing.listing_segments_json.strip()
    if raw_segments.casefold() != "runtime":
        return parse_listing_segments(raw_segments)

    parsed_override = parse_listing_segments(raw_segments)
    if parsed_override:
        return parsed_override

    runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
    brands = runtime_config.filters.include.brands
    if settings.scraping_profile.http_first and settings.listing.fast_segment_year_splits:
        segments = build_fast_listing_segments_for_makes(list(brands))
    else:
        segments = build_listing_segments_for_makes(list(brands))
    if not segments:
        logger.warning(
            "IAAI_LISTING_SEGMENTS=runtime, but runtime_config filters.brands is empty; segmented listing disabled"
        )
    return segments


@shared_task(
    name="iaai_scraper.worker.tasks.sync_segment_task",
    queue=IAAI_SYNC_QUEUE,
    bind=True,
    max_retries=2,
    default_retry_delay=60,
    acks_late=True,
)
def sync_segment_task(
    self,
    segment_index: int,
    segment: dict,
    lane: str = "iaai_cars",
    only_new: bool | None = None,
    is_bootstrap: bool = False,
):
    """Обработка одного сегмента листинга. Запускается параллельно несколькими воркерами."""
    persistence = _get_persistence()
    persistence.create_tables()
    task_id = self.request.id or "unknown"
    redis_client = _get_redis()
    settings = Settings()

    # Per-segment lock — защита от случайного дубля
    seg_lock_key = SYNC_SEGMENT_LOCK_KEY_FMT.format(idx=int(segment_index))
    owner_token = f"{task_id}:{uuid.uuid4().hex}"
    lock_ttl = _sync_segment_lock_ttl_seconds()
    lock_acquired = _acquire_lock(redis_client, seg_lock_key, owner_token, lock_ttl)
    if not lock_acquired:
        logger.info("sync_segment_task[%d] skipped: already running", segment_index)
        return {"status": "skipped", "segment_index": segment_index, "reason": "duplicate"}

    seg_make = segment.get("make")
    seg_year_min = segment.get("year_min")
    seg_year_max = segment.get("year_max")
    seg_listing_url = segment.get("listing_url")
    seg_label = f"{seg_make or 'ALL'}"
    if seg_listing_url:
        seg_label = "FILTERED_SEARCH"
    if seg_year_min is not None or seg_year_max is not None:
        seg_label += f" ({seg_year_min}-{seg_year_max})"

    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)

    _update_task_progress(
        redis_client,
        task_id=task_id,
        stage="segment_task_started",
        ttl_seconds=progress_ttl,
        segment_index=segment_index,
        segment_label=seg_label,
    )
    heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(redis_client, seg_lock_key, owner_token, lock_ttl)
    watchdog_stop, watchdog_thread = _start_stall_watchdog(
        redis_client,
        task_id=task_id,
        stall_timeout_seconds=stall_timeout,
        lock_key=seg_lock_key,
        lock_owner=owner_token,
    )

    try:
        def _job():
            with IAAIScraper() as scraper:
                scraper.set_progress_callback(
                    lambda stage, meta: _update_task_progress(
                        redis_client,
                        task_id=task_id,
                        stage=stage,
                        ttl_seconds=progress_ttl,
                        segment_index=segment_index,
                        segment_label=seg_label,
                        **meta,
                    )
                )
                return scraper.sync_listing(
                    make=seg_make,
                    model=None,
                    lane=lane,
                    only_new=only_new,
                    listing_url=str(seg_listing_url) if seg_listing_url else None,
                    year_min=seg_year_min,
                    year_max=seg_year_max,
                    skip_mark_sold=True,
                )

        result = _run_browser_job(_job)
        segment_done = bool(result.get("full_scan_completed", False))
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_completed",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
            status=result.get("status", "success"),
            cars_upserted=result.get("cars_upserted", 0),
            cars_failed=result.get("cars_failed", 0),
        )

        if is_bootstrap and segment_done:
            completed, total = _mark_segment_completed(redis_client, segment_index)
            logger.warning(
                "Segment %d (%s) bootstrap done: %d/%d completed",
                segment_index, seg_label, completed, total,
            )
            if total > 0 and completed >= total:
                _set_full_scan_done(redis_client, True)
                _clear_sync_checkpoint(redis_client)
                _clear_bootstrap_failure_streak(redis_client)
                logger.warning("All %d segments completed; bootstrap full scan done", total)

        return {
            "status": result.get("status", "success"),
            "segment_index": segment_index,
            "segment_label": seg_label,
            "cars_upserted": result.get("cars_upserted", 0),
            "cars_failed": result.get("cars_failed", 0),
            "vehicles_collected": result.get("listing", {}).get("vehicles_collected", 0),
            "full_scan_completed": segment_done,
        }

    except SoftTimeLimitExceeded:
        logger.warning("sync_segment_task[%d] soft timeout — partial progress saved", segment_index)
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_soft_timeout",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
        )
        return {
            "status": "timed_out",
            "segment_index": segment_index,
            "segment_label": seg_label,
        }
    except Exception as exc:
        logger.error("sync_segment_task[%d] failed: %s — %s", segment_index, seg_label, exc, exc_info=True)
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="segment_task_failed",
            ttl_seconds=progress_ttl,
            segment_index=segment_index,
            segment_label=seg_label,
            error=str(exc),
        )
        try:
            raise self.retry(exc=exc)
        except self.MaxRetriesExceededError:
            return {
                "status": "failed",
                "segment_index": segment_index,
                "segment_label": seg_label,
                "error": str(exc),
            }
    finally:
        if watchdog_stop is not None:
            watchdog_stop.set()
        if watchdog_thread is not None:
            watchdog_thread.join(timeout=1)
        if heartbeat_stop is not None:
            heartbeat_stop.set()
        if heartbeat_thread is not None:
            heartbeat_thread.join(timeout=1)
        _clear_task_progress(redis_client, task_id)
        _release_lock_if_owner(redis_client, seg_lock_key, owner_token)


@shared_task(
    name="iaai_scraper.worker.tasks.sync_vehicle_task",
    queue=IAAI_SYNC_QUEUE,
    bind=True,
    max_retries=2,
    default_retry_delay=30,
    acks_late=True,
)
def sync_vehicle_task(self, vehicle_url: str, lane: str = "iaai"):
    # Скрапинг и upsert одного автомобиля.
    persistence = _get_persistence()
    persistence.create_tables()

    try:
        def _job():
            with IAAIScraper() as scraper:
                return scraper.sync_vehicle(vehicle_url, lane=lane)

        result = _run_browser_job(_job)
        logger.info("sync_vehicle_task completed: %s", vehicle_url)
        return {
            "status": "success",
            "vehicle_url": vehicle_url,
            "db_action": result.get("db_action"),
            "images_upserted": result.get("images_upserted", 0),
        }

    except Exception as exc:
        logger.error("sync_vehicle_task failed: %s — %s", vehicle_url, exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name=MOBILEDE_RUNTIME_SEGMENTS_TASK,
    queue=MOBILEDE_SYNC_QUEUE,
    bind=True,
    max_retries=1,
    default_retry_delay=30,
    acks_late=True,
)
def mobilede_sync_runtime_segments_task(
    self,
    lane: str = "mobile_de_cars",
    delay_seconds: float = 0.7,
    use_cursor: bool = True,
    continuous: bool = True,
):
    redis_client = _get_redis()
    owner_token = self.request.id or uuid.uuid4().hex
    if not redis_client.set(MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY, owner_token, nx=True, ex=30 * 60):
        logger.info("mobile.de runtime segment rebuild already running")
        return {"status": "building"}
    try:
        cached_segments = _get_cached_mobilede_runtime_segments(redis_client)
        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_RUNTIME_SEGMENTS_PENDING_KEY)
            logger.info("mobile.de runtime segments rebuilt: %s", len(cached_segments))
        segments = _enqueue_mobilede_runtime_segments(
            lane=lane,
            delay_seconds=delay_seconds,
            use_cursor=use_cursor,
            continuous=continuous,
        )
        if not segments:
            logger.info("mobilede_sync_runtime_segments_task: runtime segments not configured, falling back to generic sync")
            mobilede_sync_search_task.apply_async(
                kwargs={
                    "lane": lane,
                    "delay_seconds": delay_seconds,
                    "use_cursor": use_cursor,
                    "continuous": continuous,
                },
                queue=MOBILEDE_SYNC_QUEUE,
            )
            return {"status": "fallback", "segments": 0}
        logger.info(
            "mobilede_sync_runtime_segments_task queued %s runtime segment(s): %s",
            len(segments),
            ", ".join(str(item.get("label") or item.get("make_id") or "segment") for item in segments),
        )
        return {
            "status": "queued",
            "segments": len(segments),
            "labels": [str(item.get("label") or item.get("make_id") or "segment") for item in segments],
        }
    except Exception as exc:
        logger.error("mobilede_sync_runtime_segments_task failed: %s", exc, exc_info=True)
        raise self.retry(exc=exc)
    finally:
        try:
            if redis_client.get(MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY) == owner_token:
                redis_client.delete(MOBILEDE_RUNTIME_SEGMENTS_BUILDING_KEY)
        except Exception:
            logger.debug("Failed to release mobile.de runtime segment rebuild lock", exc_info=True)


@shared_task(
    name="mobilede.sync_detail",
    queue=MOBILEDE_SYNC_QUEUE,
    bind=True,
    max_retries=2,
    default_retry_delay=30,
    acks_late=True,
)
def mobilede_sync_detail_task(self, listing_id: str, lane: str = "mobile_de_cars"):
    try:
        scraper = MobileDeScraper(persistence=_get_persistence())
        result = scraper.sync_detail(str(listing_id), lane=lane)
        logger.info("mobilede_sync_detail_task completed: %s", listing_id)
        return {"status": "success", **result}
    except Exception as exc:
        logger.error("mobilede_sync_detail_task failed: %s — %s", listing_id, exc, exc_info=True)
        raise self.retry(exc=exc)


@shared_task(
    name="mobilede.sync_search",
    queue=MOBILEDE_SYNC_QUEUE,
    bind=True,
    max_retries=2,
    default_retry_delay=60,
    acks_late=True,
)
def mobilede_sync_search_task(
    self,
    start_page: int = 1,
    max_pages: int = 5,
    lane: str = "mobile_de_cars",
    only_new: bool | None = None,
    search_url: str | None = None,
    make_id: str | None = None,
    model_id: str | None = None,
    price_min: str | None = None,
    price_max: str | None = None,
    year_min: str | None = None,
    year_max: str | None = None,
    mileage_min: str | None = None,
    mileage_max: str | None = None,
    delay_seconds: float = 0.7,
    use_cursor: bool = False,
    continuous: bool | None = None,
    segment: dict | None = None,
    segment_index: int | None = None,
    runtime_rotation: bool = False,
):
    task_id = self.request.id or "unknown"
    redis_client = _get_redis()
    settings = Settings()
    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
            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:
            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 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

    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)

    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)
        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=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)
        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),
            actual_start_page,
            actual_end_page,
            max_pages,
            use_cursor,
            continuous,
            sort_by,
            sort_order,
        )
        logger.debug(
            "mobilede_sync_search_task started: task_id=%s segment=%s start_page=%s end_page=%s max_pages=%s use_cursor=%s continuous=%s only_new=%s search_url=%s make_id=%s model_id=%s year=%s-%s price=%s-%s mileage=%s-%s",
            task_id,
            segment_label,
            actual_start_page,
            actual_end_page,
            max_pages,
            use_cursor,
            continuous,
            only_new,
            bool(search_url),
            make_id,
            model_id,
            year_min,
            year_max,
            price_min,
            price_max,
            mileage_min,
            mileage_max,
        )

        window_pages_collected = 0

        def _progress(stage: str, meta: dict[str, object]) -> None:
            nonlocal window_pages_collected
            _update_task_progress(
                redis_client,
                task_id=task_id,
                stage=stage,
                ttl_seconds=3600,
                start_page=actual_start_page,
                end_page=actual_end_page,
                use_cursor=use_cursor,
                segment_index=segment_index,
                segment_label=(segment or {}).get("label") if segment else None,
                **meta,
            )
            if stage == "search_collection_done":
                window_pages_collected = int(meta.get("pages_collected", 0) or 0)
            elif stage == "db_upsert_done":
                _log_mobilede_progress_threshold(
                    redis_client,
                    task_id=task_id,
                    segment=segment,
                    delta_pages=window_pages_collected,
                    delta_cars=int(meta.get("inserted", 0) or 0) + int(meta.get("updated", 0) or 0),
                    delta_images=int(meta.get("images_upserted", 0) or 0),
                    start_page=actual_start_page,
                    end_page=actual_end_page,
                )

        scraper = MobileDeScraper(
            client=MobileDeClient.for_worker(delay_seconds=delay_seconds),
            persistence=_get_persistence(),
        )
        result = scraper.sync_search(
            start_page=actual_start_page,
            max_pages=max_pages,
            lane=lane,
            only_new=only_new,
            sort_by=sort_by,
            sort_order=sort_order,
            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,
            progress_callback=_progress,
        )
        inserted_count = int(result.get("upsert", {}).get("inserted", 0) or 0)
        _mobilede_update_segment_freshness_state(
            redis_client,
            segment=segment,
            only_new=only_new,
            inserted=inserted_count,
            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):
            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)
        if use_cursor and listing_count == 0:
            _reset_mobilede_page_cursor(redis_client, cursor_key=cursor_key, next_start_page=1)
            logger.debug(
                "mobile.de cursor reset after empty window: task_id=%s segment=%s start_page=%s end_page=%s",
                task_id,
                (segment or {}).get("label") if segment else None,
                actual_start_page,
                actual_end_page,
            )
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="sync_done",
            ttl_seconds=3600,
            cars_upserted=result.get("upsert", {}).get("inserted", 0) + result.get("upsert", {}).get("updated", 0),
            listing_count=result.get("listing_count", 0),
            start_page=actual_start_page,
            end_page=actual_end_page,
            use_cursor=use_cursor,
            segment_index=segment_index,
            segment_label=(segment or {}).get("label") if segment else None,
        )
        logger.debug(
            "mobilede_sync_search_task completed: task_id=%s segment=%s start_page=%s end_page=%s listings=%s inserted=%s updated=%s",
            task_id,
            segment_label,
            actual_start_page,
            actual_end_page,
            result.get("listing_count"),
            result.get("upsert", {}).get("inserted", 0),
            result.get("upsert", {}).get("updated", 0),
        )
        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),
            actual_start_page,
            actual_end_page,
            int(result.get("listing_count", 0) or 0),
            int(result.get("unique_listing_count", 0) or 0),
            inserted_count,
            int(result.get("upsert", {}).get("updated", 0) or 0),
            int(result.get("upsert", {}).get("images_upserted", 0) or 0),
            result.get("run_id"),
        )
        if continuous:
            incremental_mode = bool(segment and _mobilede_bootstrap_done(redis_client) and MOBILEDE_INCREMENTAL_AFTER_BOOTSTRAP)
            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 = {
                "start_page": next_start_page,
                "max_pages": next_max_pages,
                "lane": lane,
                "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,
                "delay_seconds": delay_seconds,
                "use_cursor": use_cursor,
                "only_new": only_new,
                "continuous": True,
                "runtime_rotation": runtime_rotation,
            }
            if runtime_rotation and MOBILEDE_ROTATE_RUNTIME_SEGMENTS:
                next_segment_reservation = _reserve_next_mobilede_runtime_segment(redis_client, settings, only_new=only_new)
                if next_segment_reservation is not None:
                    next_segment_index, next_segment = next_segment_reservation
                    followup_kwargs.update(
                        {
                            "start_page": int(next_segment.get("start_page") or 1),
                            "max_pages": int(next_segment.get("max_pages") or max_pages),
                            "search_url": str(next_segment.get("search_url") or next_segment.get("listing_url") or "").strip() or None,
                            "make_id": str(next_segment.get("make_id") or "").strip() or None,
                            "model_id": str(next_segment.get("model_id") or "").strip() or None,
                            "price_min": str(next_segment.get("price_min") or "").strip() or None,
                            "price_max": str(next_segment.get("price_max") or "").strip() or None,
                            "year_min": str(next_segment.get("year_min") or "").strip() or None,
                            "year_max": str(next_segment.get("year_max") or "").strip() or None,
                            "mileage_min": str(next_segment.get("mileage_min") or "").strip() or None,
                            "mileage_max": str(next_segment.get("mileage_max") or "").strip() or None,
                            "segment": next_segment,
                            "segment_index": next_segment_index,
                        }
                    )
                    if _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
                    next_start_page = int(followup_kwargs["start_page"])
                    logger.info(
                        "mobile.de runtime rotation queued: current=%s next=%s next_pages=%s-%s",
                        segment_label,
                        _mobilede_segment_label(next_segment),
                        next_start_page,
                        next_start_page + int(followup_kwargs["max_pages"]) - 1,
                    )
            elif segment is not None and int(result.get("listing_count", 0) or 0) > 0:
                followup_kwargs["segment"] = segment
                followup_kwargs["segment_index"] = segment_index
            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,
            )
        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,
        )
    except Exception as exc:
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="failed",
            ttl_seconds=3600,
            error=str(exc),
            start_page=actual_start_page,
            end_page=actual_end_page,
            use_cursor=use_cursor,
        )
        is_transient_request_error = _is_mobilede_transient_request_error(exc)
        max_retries = int(getattr(self, "max_retries", 0) or 0)
        current_retries = int(getattr(self.request, "retries", 0) or 0)
        if is_transient_request_error:
            logger.warning(
                "mobile.de network issue: runtime=%s filter=%s pages=%s-%s retry=%s/%s error=%s",
                _mobilede_segment_label(segment),
                _mobilede_filter_source(segment, search_url),
                actual_start_page,
                actual_end_page,
                current_retries + 1,
                max_retries,
                exc,
            )
            if current_retries < max_retries:
                raise self.retry(exc=exc, countdown=max(60, MOBILEDE_CONTINUOUS_SYNC_DELAY_SECONDS))

            if continuous is None:
                continuous = MOBILEDE_CONTINUOUS_SYNC_ENABLED
            if continuous:
                followup_kwargs = {
                    "start_page": actual_start_page,
                    "max_pages": max_pages,
                    "lane": lane,
                    "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,
                    "delay_seconds": delay_seconds,
                    "use_cursor": use_cursor,
                    "continuous": True,
                }
                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),
                )
            return {
                "status": "network_error_deferred",
                "runtime": _mobilede_segment_label(segment),
                "make": _mobilede_segment_make(segment, make_id),
                "model": _mobilede_segment_model(segment, model_id),
                "pages": {
                    "start": actual_start_page,
                    "end": actual_end_page,
                    "count": actual_end_page - actual_start_page + 1,
                },
                "error": str(exc),
            }
        logger.error(
            "mobile.de sync failed: runtime=%s filter=%s pages=%s-%s error=%s",
            _mobilede_segment_label(segment),
            _mobilede_filter_source(segment, search_url),
            actual_start_page,
            actual_end_page,
            exc,
            exc_info=True,
        )
        raise self.retry(exc=exc)


@shared_task(
    name=SYNC_LISTING_TASK_NAME,
    queue=IAAI_SYNC_QUEUE,
    bind=True,
    max_retries=3,
    default_retry_delay=120,
    acks_late=True,
)
def sync_listing_task(
    self,
    make: str | None = None,
    model: str | None = None,
    lane: str = "iaai_cars",
    limit: int | None = None,
    only_new: bool | None = None,
):
    # Полный цикл: листинг + sync всех найденных машин.
    persistence = _get_persistence()
    persistence.create_tables()
    task_id = self.request.id or "unknown"
    owner_token = f"{task_id}:{uuid.uuid4().hex}"
    redis_client = _get_redis()

    lock_acquired = False
    lock_ttl = _sync_listing_lock_ttl_seconds()
    heartbeat_stop: Event | None = None
    heartbeat_thread: Thread | None = None
    watchdog_stop: Event | None = None
    watchdog_thread: Thread | None = None
    force_bootstrap_full_scan = False

    def _enqueue_bootstrap_followup(
        reason: str,
        delay_seconds: int = 5,
        *,
        count_as_failure: bool = False,
    ) -> None:
        flag_ttl = max(lock_ttl, delay_seconds + 300)
        if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
            logger.info(
                "Bootstrap follow-up already pending; skip enqueue (reason=%s)",
                reason,
            )
            return
        if count_as_failure:
            _, should_enqueue = _bump_bootstrap_failure_streak(redis_client, reason=reason)
            if not should_enqueue:
                _clear_followup_pending(redis_client)
                return
        else:
            _clear_bootstrap_failure_streak(redis_client)

        continuation_streak, should_continue = _bump_bootstrap_continuation_streak(redis_client)
        if not should_continue:
            _clear_followup_pending(redis_client)
            # Переключаемся на часовой beat-режим и останавливаем
            # немедленные bootstrap continuation, чтобы не зациклиться.
            if not always_full_scan:
                _set_full_scan_done(redis_client, True)
                _clear_sync_checkpoint(redis_client)
                logger.error(
                    "Bootstrap continuation stopped after %d immediate runs; switching to hourly schedule",
                    continuation_streak,
                )
            else:
                logger.error(
                    "Bootstrap continuation stopped after %d immediate runs (always_full_scan=true)",
                    continuation_streak,
                )
            return

        try:
            followup_expires = max(int(lock_ttl), int(delay_seconds) + 300)
            self.app.send_task(
                SYNC_LISTING_TASK_NAME,
                kwargs={
                    "make": make,
                    "model": model,
                    "lane": lane,
                    "limit": limit,
                    "only_new": only_new,
                },
                queue=IAAI_SYNC_QUEUE,
                countdown=max(0, int(delay_seconds)),
                expires=followup_expires,
            )
            logger.info(
                "Bootstrap follow-up sync queued in %ss (reason=%s)",
                delay_seconds,
                reason,
            )
        except Exception:
            _clear_followup_pending(redis_client)
            logger.warning("Failed to enqueue bootstrap follow-up sync", exc_info=True)

    lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)

    if not lock_acquired:
        orphan_cleared = _clear_orphan_sync_listing_lock(redis_client, self.app, current_task_id=task_id)
        if orphan_cleared:
            lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)

    if not lock_acquired:
        logger.info("sync_listing_task skipped: another sync is already running")
        return {
            "status": "skipped",
            "reason": "sync_already_running",
            "task_id": task_id,
        }

    try:
        settings = Settings()

        # Любой реально стартовавший sync_listing снимает pending-флаг followup,
        # чтобы watchdog/continuation могли корректно планировать следующий run
        # только при новой проблеме, а не копить дубликаты в очереди.
        _clear_followup_pending(redis_client)

        # Start heartbeat + stall watchdog immediately after lock acquisition
        # so ALL code paths (hourly, bootstrap, segmented) are protected.
        heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
            redis_client,
            SYNC_LISTING_LOCK_KEY,
            owner_token,
            lock_ttl,
            stop_on_lost=False,
        )
        stall_timeout = max(120, int(settings.celery.task_stall_timeout_seconds))
        progress_ttl = max(lock_ttl + 120, stall_timeout + 120)
        _update_task_progress(
            redis_client,
            task_id=task_id,
            stage="sync_listing_started",
            ttl_seconds=progress_ttl,
        )

        full_scan_done_before_run = _is_full_scan_done(redis_client)
        always_full_scan = bool(settings.discovery.always_full_scan)
        explicit_filtered_run = bool(
            make is not None
            or model is not None
            or (limit is not None and int(limit or 0) > 0)
            or only_new is True
        )
        force_bootstrap_full_scan = (always_full_scan or (not full_scan_done_before_run)) and not explicit_filtered_run
        if explicit_filtered_run and not full_scan_done_before_run:
            logger.info(
                "Explicit sync_listing request detected; honoring make/model/limit/only_new before bootstrap full scan is complete",
            )
        effective_limit = None if force_bootstrap_full_scan else limit
        effective_only_new = False if force_bootstrap_full_scan else only_new
        hourly_mode = settings.discovery.hourly_mode.strip().lower()
        discovery_mode = settings.discovery.mode.strip().lower()
        if always_full_scan:
            # Для режима "полный прогон каждый запуск" приоритет — устойчивый resume,
            # поэтому принудительно уходим в listing/segmented path вместо sitemap-mainline.
            discovery_mode = "listing"
        prefer_sitemap_mainline = (
            not force_bootstrap_full_scan
            and make is None
            and model is None
            and effective_limit is None
            and not effective_only_new
            and not always_full_scan
        )
        # Определяем сегменты из конфига/env или из runtime_config при IAAI_LISTING_SEGMENTS=runtime.
        filtered_listing_urls = settings.listing.filtered_search_urls
        filtered_listing_url = filtered_listing_urls[0] if filtered_listing_urls else None
        segments = _build_listing_segments(settings)

        watchdog_stop, watchdog_thread = _start_stall_watchdog(
            redis_client,
            task_id=task_id,
            stall_timeout_seconds=stall_timeout,
            lock_key=SYNC_LISTING_LOCK_KEY,
            lock_owner=owner_token,
            db_idle_restart_seconds=(DB_IDLE_RESTART_SECONDS if segments else None),
        )
        use_hourly_sitemap_sync = (
            (not always_full_scan)
            and full_scan_done_before_run
            and prefer_sitemap_mainline
            and not segments
        )

        # Segment-level checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
        # Используется только во время bootstrap для пропуска уже обработанных сегментов.
        # Никаких page-level resume — внутри сегмента всегда стартуем с page 1.
        last_completed_segment: int | None = None
        if force_bootstrap_full_scan and (always_full_scan or not full_scan_done_before_run):
            last_completed_segment = _load_last_completed_segment(redis_client)
        else:
            # После завершения bootstrap чекпоинт не нужен никогда.
            _clear_sync_checkpoint(redis_client)

        if force_bootstrap_full_scan:
            logger.info(
                "Bootstrap mode: forcing full scan (only_new=False, limit=None) until first complete run",
            )
            if always_full_scan:
                logger.info(
                    "Always full scan mode enabled (IAAI_ALWAYS_FULL_SCAN=true): using listing resume path",
                )

        logger.info(
            "sync_listing options: only_new=%s, limit=%s",
            effective_only_new,
            effective_limit,
        )

        if use_hourly_sitemap_sync:
            # Circuit breaker: если N подряд hourly-запусков фейлили, пропускаем.
            _hourly_streak, _cb_open = _check_hourly_circuit_breaker(redis_client)
            if _cb_open:
                return {
                    "status": "circuit_breaker_open",
                    "task_id": task_id,
                    "hourly_failure_streak": _hourly_streak,
                }
            _progress_cb = lambda stage, meta: _update_task_progress(
                redis_client,
                task_id=task_id,
                stage=stage,
                ttl_seconds=progress_ttl,
                **meta,
            )
            try:
                if hourly_mode == "diff":
                    logger.info("Hourly mode: running sitemap diff sync instead of full listing traversal")
                    result = _hourly_sitemap_diff_sync(
                        lane=lane,
                        limit=effective_limit,
                        only_new=effective_only_new,
                        progress_callback=_progress_cb,
                    )
                elif hourly_mode == "full_refresh":
                    logger.info("Hourly mode: running sitemap full refresh of all active vehicles")
                    result = _hourly_sitemap_full_refresh_sync(
                        lane=lane,
                        limit=effective_limit,
                        only_new=effective_only_new,
                        progress_callback=_progress_cb,
                    )
                else:
                    logger.info("Hourly mode: running sitemap rolling refresh of active vehicles")
                    result = _hourly_sitemap_rolling_refresh_sync(
                        redis_client=redis_client,
                        lane=lane,
                        limit=effective_limit,
                        only_new=effective_only_new,
                        progress_callback=_progress_cb,
                    )
            except SitemapDiscoveryError as exc:
                logger.warning(
                    "Sitemap discovery failed (%s); falling back to listing traversal for hourly sync",
                    exc,
                )
                use_hourly_sitemap_sync = False  # fall through to listing traversal below
                prefer_sitemap_mainline = False  # allow segmented fallback
                discovery_mode = "listing"  # override so use_segmented check passes

            if use_hourly_sitemap_sync:
                try:
                    redis_client.set(SITEMAP_HOURLY_LAST_COUNT_KEY, str(result.get("discovered_urls", 0)))
                except Exception:
                    logger.warning("Failed to persist hourly sitemap count", exc_info=True)

                # Anti-bot guard: при массовом protection/failed не считаем запуск успешным,
                # открываем hourly circuit breaker и уходим в controlled retry по beat.
                hourly_failures = len(result.get("failures") or [])
                hourly_discovered = int(result.get("discovered_urls") or 0)
                hourly_failed = int(result.get("cars_failed") or 0)
                hourly_protection = int(result.get("protection_events") or 0)
                if hourly_discovered > 0:
                    fail_ratio = hourly_failed / max(1, hourly_discovered)
                    protection_ratio = hourly_protection / max(1, hourly_discovered)
                    anti_bot_suspected = (
                        (hourly_protection >= 30 and protection_ratio >= 0.10)
                        or fail_ratio >= 0.30
                    )
                    if anti_bot_suspected:
                        _streak = _bump_hourly_failure_streak(redis_client)
                        logger.error(
                            "Hourly anti-bot guard triggered: discovered=%d failed=%d protection=%d fail_ratio=%.2f protection_ratio=%.2f streak=%d",
                            hourly_discovered,
                            hourly_failed,
                            hourly_protection,
                            fail_ratio,
                            protection_ratio,
                            _streak,
                        )
                        return {
                            "status": "anti_bot_detected",
                            "task_id": task_id,
                            "hourly_mode": result.get("hourly_mode"),
                            "discovered_urls": hourly_discovered,
                            "cars_failed": hourly_failed,
                            "protection_events": hourly_protection,
                            "hourly_failure_streak": _streak,
                        }

                summary = {
                    "task_id": task_id,
                    "run_id": result.get("run_id"),
                    "status": result.get("status", "success"),
                    "cars_upserted": result.get("cars_upserted", 0),
                    "cars_failed": result.get("cars_failed", 0),
                    "images_upserted": result.get("images_upserted", 0),
                    "skipped_existing": result.get("skipped_existing", 0),
                    "elapsed_seconds": result.get("elapsed_seconds"),
                    "failures_count": len(result.get("failures") or []),
                    "hourly_mode": result.get("hourly_mode"),
                    "discovered_urls": result.get("discovered_urls", 0),
                    "new_urls": result.get("new_urls", 0),
                    "sold_marked": result.get("sold_marked", 0),
                }
                logger.info(
                    "sync_listing_task hourly diff completed: status=%s, new=%d, sold=%d, skipped=%d",
                    summary["status"],
                    summary["new_urls"],
                    summary["sold_marked"],
                    summary["skipped_existing"],
                )
                # Hourly успешно — сбрасываем circuit breaker streak.
                _clear_hourly_failure_streak(redis_client)
                return summary

        _clear_followup_pending(redis_client)

        self.update_state(state="STARTED", meta={"stage": "sync_listing_started", "task_id": task_id})

        use_segmented = (
            bool(segments)
            and make is None
            and model is None
            and not use_hourly_sitemap_sync
        )

        if prefer_sitemap_mainline:
            if discovery_mode != "sitemap":
                logger.warning(
                    "Unfiltered full scan forcing sitemap discovery despite IAAI_DISCOVERY_MODE=%s",
                    discovery_mode or "unset",
                )
            if segments:
                logger.info("Ignoring configured listing segments for unfiltered sitemap full scan")

        # --- Параллельный диспатч сегментов: dispatch & exit ---
        if use_segmented and settings.celery.parallel_segments:
            # Сегменты уже завершённые (для bootstrap resume) пропускаем по Redis SET.
            already_completed: set[int] = set()
            if force_bootstrap_full_scan:
                try:
                    raw = redis_client.smembers(SYNC_SEGMENTS_PROGRESS_KEY) or set()
                    already_completed = {int(x) for x in raw if str(x).strip().lstrip("-").isdigit()}
                except Exception:
                    already_completed = set()

            pending = [
                (idx, seg) for idx, seg in enumerate(segments)
                if idx not in already_completed
            ]

            if not pending:
                # Всё уже сделано — фиксируем bootstrap done.
                if force_bootstrap_full_scan:
                    _set_full_scan_done(redis_client, True)
                    _clear_sync_checkpoint(redis_client)
                    _clear_bootstrap_failure_streak(redis_client)
                logger.warning("Parallel segments: nothing to dispatch (all completed)")
                return {
                    "status": "success",
                    "task_id": task_id,
                    "mode": "parallel_segments",
                    "segments_total": len(segments),
                    "segments_dispatched": 0,
                    "segments_already_completed": len(already_completed),
                }

            # При первом запуске bootstrap фиксируем total, чтобы знать когда остановиться.
            if force_bootstrap_full_scan and not already_completed:
                _reset_segments_progress(redis_client, len(segments))

            dispatched = 0
            for idx, seg in pending:
                try:
                    self.app.send_task(
                        "iaai_scraper.worker.tasks.sync_segment_task",
                        kwargs={
                            "segment_index": idx,
                            "segment": seg,
                            "lane": lane,
                            "only_new": effective_only_new,
                            "is_bootstrap": force_bootstrap_full_scan,
                        },
                        queue=IAAI_SYNC_QUEUE,
                    )
                    dispatched += 1
                except Exception:
                    logger.warning("Failed to dispatch segment %d", idx, exc_info=True)

            logger.warning(
                "Parallel segments dispatched: %d/%d (already_completed=%d, bootstrap=%s)",
                dispatched, len(segments), len(already_completed), force_bootstrap_full_scan,
            )
            return {
                "status": "success",
                "task_id": task_id,
                "mode": "parallel_segments",
                "segments_total": len(segments),
                "segments_dispatched": dispatched,
                "segments_already_completed": len(already_completed),
            }
        # --- конец параллельной ветки ---

        resume_from_segment = 0
        if force_bootstrap_full_scan and use_segmented and last_completed_segment is not None:
            resume_from_segment = max(0, last_completed_segment + 1)
            if resume_from_segment >= len(segments):
                # Все сегменты уже пройдены — чекпоинт устарел, начинаем заново.
                logger.info(
                    "Stored checkpoint segment=%d is beyond configured segments (%d); restarting bootstrap from segment 0",
                    last_completed_segment, len(segments),
                )
                resume_from_segment = 0
                _clear_sync_checkpoint(redis_client)
            elif resume_from_segment > 0:
                logger.warning(
                    "Resuming segmented bootstrap from segment=%d (last completed=%d)",
                    resume_from_segment, last_completed_segment,
                )

        def _progress_cb_main(stage, meta):
            _update_task_progress(
                redis_client,
                task_id=task_id,
                stage=stage,
                ttl_seconds=progress_ttl,
                **meta,
            )

        def _job():
            with IAAIScraper() as scraper:
                scraper.set_progress_callback(_progress_cb_main)
                if use_segmented:
                    return scraper.sync_listing_segmented(
                        segments=segments,
                        lane=lane,
                        only_new=effective_only_new,
                        start_segment=resume_from_segment,
                        start_page=1,
                        progress_callback=(
                            (lambda seg_idx: _save_last_completed_segment(redis_client, seg_idx))
                            if force_bootstrap_full_scan and (always_full_scan or not full_scan_done_before_run) else None
                        ),
                    )
                return scraper.sync_listing(
                    make=make,
                    model=model,
                    lane=lane,
                    limit=effective_limit,
                    only_new=effective_only_new,
                    listing_url=filtered_listing_url,
                )

        result = _run_browser_job(_job)

        if force_bootstrap_full_scan:
            bootstrap_completed = bool(result.get("full_scan_completed"))
            if bootstrap_completed:
                _set_full_scan_done(redis_client, False if always_full_scan else True)
                _clear_sync_checkpoint(redis_client)
                _clear_bootstrap_failure_streak(redis_client)
                _clear_bootstrap_continuation_streak(redis_client)
                if always_full_scan:
                    logger.info("Full scan completed; keeping bootstrap mode for next run (always full scan enabled)")
                else:
                    logger.info("Bootstrap full scan completed; hourly schedule continues")
            else:
                _set_full_scan_done(redis_client, False)
                listing_payload = result.get("listing") if isinstance(result.get("listing"), dict) else {}
                anti_bot_detected = bool(result.get("anti_bot_detected"))
                had_progress = any(
                    int(result.get(key) or 0) > 0
                    for key in ("cars_upserted", "images_upserted", "skipped_existing", "total_discovered")
                ) or int(listing_payload.get("vehicles_collected") or 0) > 0
                count_as_failure = anti_bot_detected or (str(result.get("status") or "") == "failed" and not had_progress)
                followup_delay = 5
                followup_reason = "bootstrap_not_completed"
                if anti_bot_detected:
                    followup_delay = 180
                    followup_reason = "bootstrap_anti_bot_detected"
                    logger.warning(
                        "Bootstrap anti-bot guard: protection_events=%s fail_ratio=%s protection_ratio=%s; scheduling delayed continuation",
                        result.get("protection_events"),
                        result.get("fail_ratio"),
                        result.get("protection_ratio"),
                    )
                else:
                    logger.info("Bootstrap full scan not complete yet; queuing immediate continuation")
                _enqueue_bootstrap_followup(
                    followup_reason,
                    delay_seconds=followup_delay,
                    count_as_failure=count_as_failure,
                )
        else:
            _clear_sync_checkpoint(redis_client)

        summary = {
            "task_id": task_id,
            "run_id": result.get("run_id"),
            "status": result.get("status", "success"),
            "cars_upserted": result.get("cars_upserted", 0),
            "cars_failed": result.get("cars_failed", 0),
            "protection_events": result.get("protection_events", 0),
            "images_upserted": result.get("images_upserted", 0),
            "skipped_existing": result.get("skipped_existing", 0),
            "elapsed_seconds": result.get("elapsed_seconds"),
            "failures_count": len(result.get("failures") or []),
        }
        if not force_bootstrap_full_scan:
            discovered = int(result.get("total_discovered") or result.get("total") or 0)
            failed = int(summary["cars_failed"] or 0)
            protection = int(summary["protection_events"] or 0)
            if discovered > 0:
                fail_ratio = failed / max(1, discovered)
                protection_ratio = protection / max(1, discovered)
                anti_bot_suspected = (
                    (protection >= 30 and protection_ratio >= 0.10)
                    or fail_ratio >= 0.30
                )
                if anti_bot_suspected:
                    streak = _bump_hourly_failure_streak(redis_client)
                    logger.error(
                        "Hourly anti-bot guard triggered (listing fallback): discovered=%d failed=%d protection=%d fail_ratio=%.2f protection_ratio=%.2f streak=%d",
                        discovered,
                        failed,
                        protection,
                        fail_ratio,
                        protection_ratio,
                        streak,
                    )
                    summary["status"] = "anti_bot_detected"
                    summary["hourly_failure_streak"] = streak
                    return summary
        logger.info(
            "sync_listing_task completed: status=%s, %d upserted, %d failed, failures=%d",
            summary["status"],
            summary["cars_upserted"],
            summary["cars_failed"],
            summary["failures_count"],
        )
        return summary

    except SoftTimeLimitExceeded:
        logger.warning(
            "sync_listing_task soft timeout exceeded — partial progress already saved to DB"
        )
        if force_bootstrap_full_scan:
            _set_full_scan_done(redis_client, False)
            _enqueue_bootstrap_followup("soft_time_limit_exceeded", count_as_failure=False)
        else:
            # Hourly: ставим продолжение, но только если circuit breaker не открыт.
            _bump_hourly_failure_streak(redis_client)
            _streak, _cb_open = _check_hourly_circuit_breaker(redis_client)
            if not _cb_open:
                followup_ttl = max(600, int(lock_ttl))
                if _try_set_followup_pending(redis_client, ttl_seconds=followup_ttl):
                    try:
                        self.app.send_task(
                            SYNC_LISTING_TASK_NAME,
                            kwargs={
                                "make": make,
                                "model": model,
                                "lane": lane,
                                "limit": limit,
                                "only_new": only_new,
                            },
                            queue=IAAI_SYNC_QUEUE,
                            countdown=10,
                            expires=followup_ttl,
                        )
                        logger.info("Queued immediate continuation after soft timeout")
                    except Exception:
                        _clear_followup_pending(redis_client)
                        logger.warning("Failed to queue continuation after soft timeout", exc_info=True)
                else:
                    logger.info("Continuation after soft timeout already pending; skip duplicate enqueue")
            else:
                logger.warning("Skipping continuation: hourly circuit breaker open (%d failures)", _streak)
        return {
            "status": "timed_out",
            "task_id": task_id,
            "reason": "soft_time_limit_exceeded",
            "note": "partial progress saved to DB; continuation queued",
        }

    except Exception as exc:
        logger.error("sync_listing_task failed: %s", exc, exc_info=True)
        # Hourly circuit breaker: фиксируем ошибку.
        if not force_bootstrap_full_scan:
            _bump_hourly_failure_streak(redis_client)
        try:
            if force_bootstrap_full_scan:
                _set_full_scan_done(redis_client, False)
                raise self.retry(exc=exc, countdown=5)
            raise self.retry(exc=exc)
        except self.MaxRetriesExceededError:
            logger.error("sync_listing_task max retries exceeded, giving up")
            if force_bootstrap_full_scan:
                _set_full_scan_done(redis_client, False)
                _enqueue_bootstrap_followup("max_retries_exceeded", count_as_failure=True)
            # Non-bootstrap: НЕ ставим continuation — beat поставит новую задачу
            # через beat_sync_interval_minutes. Бесконечный retry при ошибках
            # приводит к молотилке запросов и бану.
            return {
                "status": "failed",
                "task_id": task_id,
                "error": str(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)))
        if lock_acquired:
            _release_lock_if_owner(redis_client, SYNC_LISTING_LOCK_KEY, owner_token)